fix(retention): accumulate late-arriving spans instead of overwriting a rolled-up day

daily_usage was written with INSERT OR REPLACE, which is correct only while a
roll-up recomputes a day from complete raw data. A span dated to a day that had
already been rolled up and purged — a backfill, or an import of telemetry older
than COTEL_RETENTION_RAW_DAYS — made the next cycle see exactly one span for
that day, recompute the aggregate from it alone, and REPLACE the correct row.
The day's earlier usage was gone, and so were the raw spans that could rebuild
it, so historical cost was silently understated.

A day's aggregate now accumulates: ON CONFLICT DO UPDATE adds each cycle's sum
to the existing row instead of replacing it.

This reverses the reasoning recorded in the previous commit, which kept
INSERT OR REPLACE because an accumulating upsert would double a day's aggregate
on a retry after a crash between the INSERT and the DELETE. That objection was
correct and is answered rather than ignored: the accumulate, the raw-span purge
and the aggregate purge now run in a single transaction, so the spans a cycle
folds in vanish atomically with the addition and a crash rolls both back. The
whole-day truncation from that commit still stands — it is what keeps a day
from being rolled up in slices in the first place.

Accumulation is safe against the trigger it exists for: spans.span_id is a
PRIMARY KEY and both ingest and import insert with OR IGNORE, so re-importing a
span is a no-op and cannot inflate a total. The COALESCE on the update side
covers total_cache_read_tokens and total_cache_write_tokens only, because those
are the columns the v7->v8 migration added nullable; the other totals are
non-pointer types on every write path and are never NULL.

Tests roll up a day, ingest a span dated into that same purged day, and assert
the aggregate is the sum of both spans, that the day keeps exactly one row, and
that a further cycle does not double-count.

Co-Authored-By: Daedalus <daedalus@agents.flopbut.local>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Daedalus

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - No more silent telemetry loss on restart/deploy: the ingest (`:4318`) and dashboard (`:8080`) ports now bind and accept connections **before** the DuckDB open (WAL replay + schema migration), which can take minutes on a large production database. During that window both ports answer a retryable `503` with `Retry-After` instead of resetting the connection, so OTLP exporters retry and spans are delivered once cotel is ready. Previously the ports bound only after the open finished, so every deploy had a multi-minute window where ingest refused connections and dropped spans
 - Retention roll-up no longer loses part of a day's usage. The cutoff was a wall-clock instant, so with the worker ticking every 6h (`COTEL_RETENTION_INTERVAL`) a day was rolled up in slices; because `daily_usage` is keyed by day and written with `INSERT OR REPLACE`, each slice overwrote the previous one while its raw spans were already purged. Every day's `span_count`, token totals and `total_cost_usd` were therefore systematically understated — only the last slice survived. The cutoff is now snapped back to UTC midnight — the same boundary the `day` key is bucketed on — so only whole days are ever rolled up and purged. Trade-off: raw spans now live up to one day longer than `COTEL_RETENTION_RAW_DAYS` (default 30). Aggregates already flattened by the old behaviour cannot be recovered — the raw spans are gone
 - Retention is now correct on servers that do not run in UTC. A span's `day` is its UTC calendar day, but both retention cutoffs were computed in the server's local zone, so on any host at a non-zero UTC offset — including UTC+1/+2 — the boundary fell inside a day bucket and re-introduced the overwrite above for spans near midnight UTC. Both cutoffs are now computed in UTC, and the retention tests run under a matrix of server timezones so the alignment cannot silently regress
+- Retention roll-up no longer overwrites an already-rolled-up day's aggregate when a span dated to that day arrives late — a backfill, or an import of telemetry older than `COTEL_RETENTION_RAW_DAYS`. The day had already been aggregated and its raw spans purged, so the next cycle recomputed the day from the single late span alone and `INSERT OR REPLACE`d the correct total away, silently corrupting historical cost. Late spans now **accumulate** into the existing `daily_usage` row (`ON CONFLICT DO UPDATE`) instead of replacing it; the accumulate and the raw-span purge run in one transaction so a crash between them cannot double-count
 
 ### Added
 - Startup logging of each phase — `opening db`, `db ready: schema/migrations applied in <duration>`, and `ready: serving live traffic …` — so a slow open is visible instead of a silent container. Plus a Docker `HEALTHCHECK` (`cotel -healthcheck`, which probes the dashboard `/healthz`) so the container reports `health: starting` until the database is open rather than a misleading `Up`; no curl/wget is added to the runtime image

README.md

@@ -210,10 +210,8 @@ Override with environment variables:
 
 The worker ticks several times a day, but it only ever rolls up and purges
 **complete** calendar days: the raw-span cutoff is snapped back to midnight
-before use. `daily_usage` is keyed by day and rewritten with `INSERT OR REPLACE`,
-so rolling up a day in slices would make each slice overwrite the one before it
-while its raw spans were already deleted — permanently understating that day's
-`span_count`, tokens and cost.
+before use. Rolling up a day in slices would risk a slice's raw spans being
+purged before the rest of the day is aggregated.
 
 Those are **UTC** calendar days, and the cutoff is UTC midnight, whatever
 timezone the server itself runs in. Aggregates are bucketed by UTC day
@@ -222,6 +220,16 @@ everywhere, so daily figures do not shift with the host's zone.
 The practical effect: a raw span survives up to one day longer than
 `COTEL_RETENTION_RAW_DAYS` before it is aggregated away.
 
+### Late-arriving spans accumulate
+
+A day's aggregate is built by **accumulation**: each roll-up cycle *adds* its sum
+to the existing `daily_usage` row (`ON CONFLICT DO UPDATE`) rather than replacing
+it. So a span dated to a day that was already rolled up and purged — a backfill,
+or an import (`POST /api/v1/import`) of telemetry older than
+`COTEL_RETENTION_RAW_DAYS` — is added to that day's total instead of overwriting
+it with itself alone. The accumulate and the raw-span purge run in a single
+transaction, so a crash between them cannot double-count.
+
 ### Unattributed usage — the `unknown` sentinel
 
 `daily_usage` is keyed by `(day, session_id, model, tool_name)`. A raw span that

internal/storage/retention.go

@@ -94,12 +94,16 @@ func (db *DB) rollupAndPurgeAt(cfg RetentionConfig, now time.Time) error {
 	// Roll raw spans older than RawDays into daily_usage before deleting.
 	//
 	// The cutoff is snapped back to midnight so a roll-up only ever consumes
-	// whole days. daily_usage is keyed by day and written with INSERT OR
-	// REPLACE, so a cutoff landing mid-day would recompute that day from
-	// whatever raw spans the previous cycle left behind and overwrite the
-	// earlier slice — whose spans are already deleted. Whole days keep REPLACE
-	// correct and the whole operation idempotent, at the cost of spans living
-	// up to one day longer than RawDays.
+	// whole days: a span survives up to one day longer than RawDays but a day is
+	// never rolled up in slices.
+	//
+	// A day's aggregate accumulates: ON CONFLICT DO UPDATE adds each cycle's sum
+	// to the existing row rather than replacing it. A span dated to a day that was
+	// already rolled up and purged — a backfill or an import of old telemetry —
+	// therefore adds to that day's total instead of overwriting it with itself
+	// alone. The accumulate and the raw-span DELETE must stay in one transaction:
+	// the spans a cycle folds in have to be gone atomically with the addition, or
+	// a crash between them lets a retry count them twice.
 	//
 	// That midnight is UTC, never the server's local midnight: the day column
 	// below comes from CAST(start_time AS TIMESTAMP), which renders the stored
@@ -114,8 +118,16 @@ func (db *DB) rollupAndPurgeAt(cfg RetentionConfig, now time.Time) error {
 	// normalised values drive the grouping — '' and NULL collapse into one
 	// 'unknown' bucket rather than two.
 	rollupCutoff := startOfDayUTC(now.AddDate(0, 0, -cfg.RawDays))
-	_, err := db.rw.Exec(`
-		INSERT OR REPLACE INTO daily_usage
+	aggCutoff := startOfDayUTC(now.AddDate(0, 0, -cfg.AggregateDays))
+
+	tx, err := db.rw.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	if _, err := tx.Exec(`
+		INSERT INTO daily_usage
 		  (day, session_id, model, tool_name, user_id,
 		   span_count, total_input_tokens, total_output_tokens,
 		   total_cache_read_tokens, total_cache_write_tokens, total_cost_usd)
@@ -134,21 +146,31 @@ func (db *DB) rollupAndPurgeAt(cfg RetentionConfig, now time.Time) error {
 		FROM spans
 		WHERE start_time < ?
 		GROUP BY 1, 2, 3, 4
-	`, UnknownSentinel, UnknownSentinel, UnknownSentinel, rollupCutoff)
-	if err != nil {
+		ON CONFLICT (day, session_id, model, tool_name) DO UPDATE SET
+		  span_count               = daily_usage.span_count + excluded.span_count,
+		  total_input_tokens       = daily_usage.total_input_tokens + excluded.total_input_tokens,
+		  total_output_tokens      = daily_usage.total_output_tokens + excluded.total_output_tokens,
+		  total_cache_read_tokens  = COALESCE(daily_usage.total_cache_read_tokens, 0) + excluded.total_cache_read_tokens,
+		  total_cache_write_tokens = COALESCE(daily_usage.total_cache_write_tokens, 0) + excluded.total_cache_write_tokens,
+		  total_cost_usd           = daily_usage.total_cost_usd + excluded.total_cost_usd,
+		  user_id                  = COALESCE(daily_usage.user_id, excluded.user_id)
+	`, UnknownSentinel, UnknownSentinel, UnknownSentinel, rollupCutoff); err != nil {
 		return err
 	}
 
-	// Purge raw spans past RawDays.
-	if _, err := db.rw.Exec(`DELETE FROM spans WHERE start_time < ?`, rollupCutoff); err != nil {
+	// Purge raw spans past RawDays — atomic with the accumulate above.
+	if _, err := tx.Exec(`DELETE FROM spans WHERE start_time < ?`, rollupCutoff); err != nil {
 		return err
 	}
 
 	// Purge aggregates past AggregateDays. Also UTC-snapped: day is a DATE, so
 	// a local-zone instant here would round to a different day than the one the
 	// roll-up wrote and drop (or keep) a day's aggregate a day early or late.
-	aggCutoff := startOfDayUTC(now.AddDate(0, 0, -cfg.AggregateDays))
-	if _, err := db.rw.Exec(`DELETE FROM daily_usage WHERE day < ?`, aggCutoff); err != nil {
+	if _, err := tx.Exec(`DELETE FROM daily_usage WHERE day < ?`, aggCutoff); err != nil {
+		return err
+	}
+
+	if err := tx.Commit(); err != nil {
 		return err
 	}
 

internal/storage/retention_test.go

@@ -284,6 +284,64 @@ func TestRollupAndPurge_Idempotent(t *testing.T) {
 	})
 }
 
+// TestRollupAndPurge_LateSpanAccumulates pins the fix for a late-arriving span:
+// a span whose day was already rolled up and purged, ingested afterwards (a
+// backfill or an import of old telemetry), must be added to that day's existing
+// aggregate on the next cycle — not REPLACE it. Before the fix the second
+// roll-up recomputed the day from the single late span alone and overwrote the
+// correct total, silently corrupting historical cost.
+func TestRollupAndPurge_LateSpanAccumulates(t *testing.T) {
+	db, err := Open(":memory:")
+	if err != nil {
+		t.Fatalf("open in-memory db: %v", err)
+	}
+	defer db.Close()
+
+	day := time.Date(2026, 1, 15, 0, 0, 0, 0, time.Local)
+	insertBoundarySpan(t, db, "first", day.Add(2*time.Hour), 100, 10, 1000, 200, 0.10)
+
+	const rawDays = 30
+	cfg := RetentionConfig{RawDays: rawDays, AggregateDays: 90}
+	tick := day.AddDate(0, 0, rawDays+1).Add(8 * time.Hour)
+
+	// Roll up the day; its only raw span is aggregated then purged.
+	if err := db.rollupAndPurgeAt(cfg, tick); err != nil {
+		t.Fatalf("first roll-up: %v", err)
+	}
+
+	// A late span dated to the same already-rolled-up day arrives afterwards,
+	// sharing the aggregate's PK (same session/model/tool).
+	insertBoundarySpan(t, db, "late", day.Add(10*time.Hour), 7, 3, 70, 30, 0.01)
+
+	if err := db.rollupAndPurgeAt(cfg, tick); err != nil {
+		t.Fatalf("second roll-up (late span): %v", err)
+	}
+
+	// The aggregate must be the sum of both spans, not just the late one.
+	want := dayTotals{spans: 2, input: 107, output: 13, cacheRead: 1070, cacheWrite: 230, cost: 0.11}
+	if got := accountedFor(t, db, day); got != want {
+		t.Errorf("late span overwrote the day: accounted %+v, want %+v", got, want)
+	}
+
+	// Accumulation must not fork a second aggregate row for the day.
+	var rows int64
+	if err := db.rw.QueryRow(`SELECT COUNT(*) FROM daily_usage WHERE day = CAST(? AS DATE)`,
+		day.Format("2006-01-02")).Scan(&rows); err != nil {
+		t.Fatalf("count daily_usage rows for day: %v", err)
+	}
+	if rows != 1 {
+		t.Errorf("daily_usage rows for the day: got %d, want 1", rows)
+	}
+
+	// With the late span now purged, re-running must not double-count it.
+	if err := db.rollupAndPurgeAt(cfg, tick); err != nil {
+		t.Fatalf("idempotent re-run: %v", err)
+	}
+	if got := accountedFor(t, db, day); got != want {
+		t.Errorf("re-run changed the aggregate: accounted %+v, want %+v", got, want)
+	}
+}
+
 type dayTotals struct {
 	spans                 int64
 	input, output         int64