Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 52 additions & 2 deletions src/services/metrics.ts
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,48 @@ export class Timer {
}
}

/**
* GIT-73: session rows still being created, by session id.
*
* session_start creates its session row fire-and-forget and records its own
* metrics at the same moment, and gitmem_query_metrics.session_id references
* gitmem_sessions(id). Whichever request reached the store first decided
* whether the metrics insert 409'd (FK violation). recall right after
* session_start raced the same way.
*
* A metrics write for a session registered here waits for its row. If the row
* did not land (or takes longer than the wait budget), the metrics row is
* written without the FK and keeps the id in metadata.session_id — where
* session_close already looks for it (GIT-109) — instead of being lost to a 409.
* Entries stay after they settle so later writes know the answer too.
*/
const pendingSessionRows = new Map<string, Promise<boolean>>();
const SESSION_ROW_WAIT_MS = 15_000;

export function registerPendingSessionRow(sessionId: string, created: Promise<boolean>): void {
pendingSessionRows.set(sessionId, created.catch(() => false));
}

/** For tests. */
export function clearPendingSessionRows(): void {
pendingSessionRows.clear();
}

async function sessionRowLanded(sessionId: string): Promise<boolean | undefined> {
const pending = pendingSessionRows.get(sessionId);
if (!pending) return undefined; // not created by this process — nothing to wait for
let timer: ReturnType<typeof setTimeout> | undefined;
const budget = new Promise<boolean>((resolve) => {
timer = setTimeout(() => resolve(false), SESSION_ROW_WAIT_MS);
timer.unref?.();
});
try {
return await Promise.race([pending, budget]);
} finally {
if (timer) clearTimeout(timer);
}
}

/**
* Track a query and record metrics.
*
Expand All @@ -146,9 +188,17 @@ export async function recordMetrics(metrics: QueryMetrics): Promise<void> {
// Auto-detect agent if not provided by caller
const agent = metrics.agent || getAgentIdentity() || null;

// GIT-73: never reference a session row that is not there yet.
let sessionId = metrics.session_id || null;
let metadata = metrics.metadata || {};
if (sessionId && (await sessionRowLanded(sessionId)) === false) {
metadata = { ...metadata, session_id: sessionId };
sessionId = null;
}

const record: Record<string, unknown> = {
id: metrics.id,
session_id: metrics.session_id || null,
session_id: sessionId,
agent,
tool_name: metrics.tool_name,
query_text: metrics.query_text || null,
Expand All @@ -160,7 +210,7 @@ export async function recordMetrics(metrics: QueryMetrics): Promise<void> {
phase_tag: metrics.phase_tag || null,
linear_issue: metrics.linear_issue || null,
memories_surfaced: metrics.memories_surfaced || null,
metadata: metrics.metadata || {},
metadata,
created_at: new Date().toISOString(),
};

Expand Down
15 changes: 11 additions & 4 deletions src/tools/session-start.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import { getStorage } from "../services/storage.js";
import {
Timer,
recordMetrics,
registerPendingSessionRow,
calculateContextBytes,
PERFORMANCE_TARGETS,
buildPerformanceData,
Expand Down Expand Up @@ -426,14 +427,14 @@ async function createSessionRecord(
project: Project,
linearIssue?: string,
preGeneratedId?: string // Accept pre-generated UUID for fire-and-forget pattern
): Promise<{ session_id: string; latency_ms: number; network_call: boolean }> {
): Promise<{ session_id: string; latency_ms: number; network_call: boolean; stored: boolean }> {
const sessionId = preGeneratedId || uuidv4();
const today = new Date().toISOString().split("T")[0];
const timer = new Timer();

if (!hasSupabase()) {
// Free tier: session tracked locally only
return { session_id: sessionId, latency_ms: timer.stop(), network_call: false };
return { session_id: sessionId, latency_ms: timer.stop(), network_call: false, stored: false };
}

try {
Expand All @@ -458,6 +459,7 @@ async function createSessionRecord(
session_id: sessionId,
latency_ms: timer.stop(),
network_call: true, // Always writes to Supabase
stored: true,
};
} catch (error) {
console.error("[session_start] Failed to create session record:", error);
Expand All @@ -466,6 +468,7 @@ async function createSessionRecord(
session_id: sessionId,
latency_ms: timer.stop(),
network_call: true, // Network was attempted
stored: false,
};
}
}
Expand Down Expand Up @@ -1117,8 +1120,12 @@ export async function sessionStart(
console.error(`[session_start] Resuming session ${sessionId} — skipping record creation`);
} else {
sessionId = uuidv4();
// Fire-and-forget: don't await the Supabase write
createSessionRecord(agent, project, params.linear_issue, sessionId).catch(() => {});
// Fire-and-forget: don't await the Supabase write. GIT-73: metrics writes
// for this session wait on it instead of racing it to a 409.
registerPendingSessionRow(
sessionId,
createSessionRecord(agent, project, params.linear_issue, sessionId).then((r) => r.stored)
);
// Mark prior in-memory session as superseded (force=true path)
// Registry displacement in writeSessionFiles handles the registry case,
// but priorSession may not be in the registry (e.g., after MCP restart)
Expand Down
3 changes: 1 addition & 2 deletions tests/e2e/blank-supabase.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ const DENY_REFS = new Set(["cjptxyezuxdiinufgrrm"]); // production GitMem — ne
* Known venue failures. Any other 4xx/5xx from /rest/v1/ or /functions/v1/
* fails the run. Each entry names the ticket that removes it — delete the entry
* when that ticket merges, so this list only shrinks. An entry allows a failure;
* it does not require one (e.g. the GIT-73 race is intermittent).
* it does not require one (a failure may be intermittent).
*/
/**
* Not defects: the store capability probe (src/services/store-columns.ts) is a
Expand All @@ -72,7 +72,6 @@ function isCapabilityProbe(r) {
}

const EXPECTED_FAILURES = [
{ ticket: "GIT-73", what: "metrics/session_start FK race", method: "POST", path: /^\/rest\/v1\/gitmem_query_metrics$/, status: 409 },
{ ticket: "GIT-105", what: "knowledge-triple thread id into a uuid column", method: "POST", path: /^\/rest\/v1\/knowledge_triples$/, status: 400 },
];
const HERE = dirname(fileURLToPath(import.meta.url));
Expand Down
74 changes: 74 additions & 0 deletions tests/unit/services/metrics-session-fk.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
/**
* GIT-73: a metrics row never references a session row that is not there yet.
*
* session_start created its session row fire-and-forget and wrote its metrics
* row at the same moment; gitmem_query_metrics.session_id is an FK to
* gitmem_sessions(id), so whichever arrived first decided whether the metrics
* insert 409'd.
*/
import { describe, it, expect, vi, beforeEach } from "vitest";

const writes = vi.hoisted(() => ({ log: [] as Array<{ table: string; row: Record<string, unknown>; at: number }> }));

vi.mock("../../../src/services/supabase-client.js", () => ({
directUpsert: vi.fn(async (table: string, row: Record<string, unknown>) => {
writes.log.push({ table, row, at: Date.now() });
return row;
}),
}));
vi.mock("../../../src/services/tier.js", async (orig) => ({
...(await orig<typeof import("../../../src/services/tier.js")>()),
hasSupabase: () => true,
}));

import { recordMetrics, registerPendingSessionRow, clearPendingSessionRows } from "../../../src/services/metrics.js";

const SID = "393adb34-a80c-4c3a-b71a-bc0053b7a7ea";
const metric = (extra: Record<string, unknown> = {}) => ({
id: "m-" + Math.random().toString(16).slice(2), session_id: SID, tool_name: "session_start",
latency_ms: 1, result_count: 0, metadata: { project: "p" }, ...extra,
}) as never;

beforeEach(() => { writes.log = []; clearPendingSessionRows(); });

describe("metrics wait for their session row (GIT-73)", () => {
it("waits until the session row has landed, then keeps the FK", async () => {
let land!: (v: boolean) => void;
registerPendingSessionRow(SID, new Promise<boolean>((r) => (land = r)));

const pending = recordMetrics(metric());
await new Promise((r) => setTimeout(r, 30));
expect(writes.log).toHaveLength(0); // held back while the session row is in flight

land(true);
await pending;
expect(writes.log).toHaveLength(1);
expect(writes.log[0].row.session_id).toBe(SID);
});

it("a session row that did not land: no FK, id kept in metadata.session_id", async () => {
registerPendingSessionRow(SID, Promise.resolve(false));
await recordMetrics(metric());
expect(writes.log[0].row.session_id).toBeNull();
expect(writes.log[0].row.metadata).toEqual({ project: "p", session_id: SID });
});

it("a session row whose creation threw is treated as not landed", async () => {
registerPendingSessionRow(SID, Promise.reject(new Error("store down")));
await recordMetrics(metric());
expect(writes.log[0].row.session_id).toBeNull();
});

it("later writes for the same session reuse the settled answer", async () => {
registerPendingSessionRow(SID, Promise.resolve(true));
await recordMetrics(metric());
await recordMetrics(metric({ tool_name: "recall" }));
expect(writes.log.map((w) => w.row.session_id)).toEqual([SID, SID]);
});

it("a session this process did not create (resumed, or another process's) is written immediately", async () => {
await recordMetrics(metric());
expect(writes.log).toHaveLength(1);
expect(writes.log[0].row.session_id).toBe(SID);
});
});
Loading