Keep a span's created_at at its start when re-persisting

Batched writes re-send a span row on every flush it's dirty for (open,
attribute merges, close), and `persist` used INSERT OR REPLACE, which
deletes the row before re-inserting. That re-stamped the `created_at`
default with the time of the latest flush. The trace list renders the
root span's `created_at`, and the root is re-sent when the invocation
closes, so the displayed time showed when a request finished rather
than when it started.

Upsert instead, leaving `created_at` alone — matching what the
write-through path did with INSERT ... DO NOTHING followed by UPDATE.

Also correct the FLUSH_INTERVAL_MS comment: closing spans are buffered
like any other row, so only logs and exceptions are written as they
arrive.

Author: nickpatt

packages/miniflare/src/workers/observability/tail-to-store.ts

@@ -38,8 +38,9 @@ const FLUSH_THRESHOLD = 16;
  * Time comes from tail-event timestamps, not `Date.now()`, which a Worker only
  * advances on I/O. So this bounds staleness *between events*, not in wall-clock
  * time: an invocation that goes completely quiet flushes nothing further until
- * its outcome. Logs and closing spans are written as they happen, which is what
- * covers the quiet case in practice.
+ * its outcome. Logs and exceptions are written as they arrive, so a quiet
+ * invocation can still report what it's doing; a closing span is buffered like
+ * any other row, so its duration can trail the close event by one flush.
  */
 const FLUSH_INTERVAL_MS = 100;
 

packages/miniflare/src/workers/observability/trace-store.ts

@@ -103,10 +103,25 @@ export class TraceStore extends DurableObject {
 	/** Persist one invocation's spans + logs. Called by the collector. */
 	persist(spans: SpanInput[], logs: LogInput[]): void {
 		for (const s of spans) {
+			// Upsert rather than INSERT OR REPLACE: a span is re-sent on every flush
+			// it's dirty for (open, attribute merges, close), and REPLACE deletes the
+			// row first, so `created_at` would be re-stamped with the latest flush.
+			// The trace list renders the root span's `created_at`, which would then
+			// show when the invocation finished rather than when it started.
 			this.sql.exec(
-				`INSERT OR REPLACE INTO spans
+				`INSERT INTO spans
 					(trace_id, span_id, parent_id, service, name, kind, start_ms, duration_ms, outcome, error, attributes)
-					VALUES (?,?,?,?,?,?,?,?,?,?, jsonb(?))`,
+					VALUES (?,?,?,?,?,?,?,?,?,?, jsonb(?))
+					ON CONFLICT (trace_id, span_id) DO UPDATE SET
+						parent_id = excluded.parent_id,
+						service = excluded.service,
+						name = excluded.name,
+						kind = excluded.kind,
+						start_ms = excluded.start_ms,
+						duration_ms = excluded.duration_ms,
+						outcome = excluded.outcome,
+						error = excluded.error,
+						attributes = excluded.attributes`,
 				s.traceId,
 				s.spanId,
 				s.parentId,

packages/miniflare/test/plugins/core/observability.spec.ts

@@ -179,6 +179,19 @@ export default {
 			});
 			return new Response("closed");
 		}
+		if (url.pathname === "/repersist") {
+			const store = env.TRACE_STORE.get(env.TRACE_STORE.idFromName("singleton"));
+			// The same span re-sent as a later batch flush would send it: still
+			// running on the first call, closed on the second.
+			await store.persist([{
+				traceId: "trace-re", spanId: "root", parentId: null,
+				name: "agent run", kind: "http", startMs: 7000,
+				durationMs: url.searchParams.get("close") ? 2500 : null,
+				outcome: url.searchParams.get("close") ? "ok" : null,
+				error: null, attributes: { "faas.trigger": "http" },
+			}], []);
+			return new Response("repersisted");
+		}
 		if (url.pathname.startsWith("/wobs/")) {
 			return env.WOBS.fetch(
 				new Request("http://collector" + url.pathname.slice("/wobs".length) + url.search, request)
@@ -370,6 +383,54 @@ describe("unsafeObservability (write-through capture)", () => {
 		expect(rootClosed.duration_ms).toBe(1234);
 		expect(JSON.parse(rootClosed.attributes ?? "{}")["cpu_time_ms"]).toBe(2);
 	});
+
+	test("re-persisting a span updates it without re-stamping created_at", async ({
+		expect,
+	}) => {
+		const mf = new Miniflare({
+			unsafeObservability: true,
+			workers: [storeWorker()],
+		});
+		useDispose(mf);
+
+		async function readRe() {
+			const rows = await queryStore(
+				mf,
+				`SELECT duration_ms, outcome, created_at FROM spans
+					WHERE trace_id = ? AND span_id = 'root'`,
+				["trace-re"]
+			);
+			assert(rows[0], "expected the re-persisted span");
+			return rows[0] as {
+				duration_ms: number | null;
+				outcome: string | null;
+				created_at: string;
+			};
+		}
+
+		expect(
+			await (await mf.dispatchFetch("http://localhost/repersist")).text()
+		).toBe("repersisted");
+		const open = await readRe();
+		expect(open.duration_ms).toBe(null);
+
+		// `created_at` defaults to `datetime('now')`, which has whole-second
+		// granularity, so wait long enough that a re-stamp would be visible.
+		await new Promise((resolve) => setTimeout(resolve, 1100));
+
+		expect(
+			await (
+				await mf.dispatchFetch("http://localhost/repersist?close=1")
+			).text()
+		).toBe("repersisted");
+		const closed = await readRe();
+		// The update landed...
+		expect(closed.duration_ms).toBe(2500);
+		expect(closed.outcome).toBe("ok");
+		// ...but the row was upserted, not deleted and re-inserted, so the trace
+		// list still shows when the invocation started rather than when it ended.
+		expect(closed.created_at).toBe(open.created_at);
+	});
 });
 
 // A plain user worker (no manual seeding): it does some work, logs, and proxies