sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(agent): list SQLite sessions without writer claims

Browse source
This commit is contained in:
Christian Klotz 2026-08-05 12:35:21 +03:00
parent 7cf90c1d18
commit e13e1f7667
No known key found for this signature in database
12 changed files with 229 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/agent/docs/harness-v2.md
View file

@ -1683,7 +1683,7 @@ lane_moves (session_id, seq, lane, leaf_id) -- history; getLog parity
facts (session_id, seq, kind, key, value) -- name, labels; latest by seq
branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
branch_tips (session_id, branch_id, tip_id) -- PRIMARY KEY (session_id, tip_id)
leases (session_id, owner_id, fence, expires_at_ms) -- writer claim
writer_leases (session_id, owner_id, fence, expires_at_ms) -- writer claim
-- indexes
records: (session_id, lane, type, seq), (session_id, lane, type, op_kind, seq)
@ -1692,7 +1692,9 @@ branch_entries: (session_id, branch_id, entry_type, entry_seq)
(session_id, entry_id) -- reverse lookup: entry → branches
```
`leases` enforces one writer per session with expiring, fenced claims. Storage renews the claim inside every write transaction and while idle. Repository-owned cleanup releases only its matching owner and fence.
`writer_leases` enforces one writer per session with expiring, fenced claims. Storage renews the claim inside every write transaction and while idle. Repository-owned cleanup releases only its matching owner and fence.
`open()` acquires that writer claim. `list()` never acquires or renews writer leases: it reads every matching session directly from the session catalog and projects the latest name fact into the top-level `SqliteSessionMetadata.name` field for server-side inventory. Application-owned `SqliteSessionMetadata.metadata` remains unchanged.
`branch_entries` and `branch_tips` are a private read cache. No interface exposes them; no other backend has them; rebuilding them from parent pointers is an explicit repair operation, never a runtime fallback.

2
packages/agent/src/harness/session/types.ts
View file

@ -337,7 +337,9 @@ export interface SessionRepo<
TListOptions = void,
> {
create(options: TCreateOptions): Promise<Session<TMetadata>>;
/** Opens the session for writing and acquires any backend writer claim. */
open(metadata: TMetadata): Promise<Session<TMetadata>>;
/** Lists session metadata without opening sessions or acquiring writer claims. */
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(source: TMetadata, options: ForkOptions & TCreateOptions): Promise<Session<TMetadata>>;

2
packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql
View file

@ -117,7 +117,7 @@ CREATE TABLE IF NOT EXISTS branch_tips (
-- Per-session writer claim. The fence prevents an expired owner from writing
-- after a new owner takes over the session.
CREATE TABLE IF NOT EXISTS leases (
CREATE TABLE IF NOT EXISTS writer_leases (
session_id TEXT PRIMARY KEY,
owner_id TEXT NOT NULL,
fence INTEGER NOT NULL,

67
packages/session-backends/sqlite-node/src/sqlite/repo.ts
View file

@ -43,13 +43,6 @@ import {
setLaneLeaf,
moveLane as updateLane,
} from "./storage/lanes.ts";
import {
acquireSessionLease,
deleteSessionLease,
releaseSessionLease,
renewSessionLease,
type SessionLease,
} from "./storage/leases.ts";
import {
appendRecordRow,
deleteRecordRows,
@ -72,14 +65,21 @@ import {
readStats,
} from "./storage/session-stats.ts";
import {
decodeSessionMetadata,
deleteSessionRow,
insertSessionRow,
readSessionRow,
readSessionRows,
rowToMetadata,
type SessionRow,
sessionExists,
} from "./storage/sessions.ts";
import {
acquireWriterLease,
deleteWriterLease,
releaseWriterLease,
renewWriterLease,
type WriterLease,
} from "./storage/writer-leases.ts";
import type {
SqliteDatabase,
SqliteDatabaseFactory,
@ -126,9 +126,9 @@ function lostWriterError(sessionId: string): SessionError {
return new SessionError("storage", `SQLite session ${sessionId} writer lease was lost`);
}
function acquireWriterLease(db: SqliteDatabase, sessionId: string, options: ResolvedWriterLeaseOptions): SessionLease {
function claimWriterLease(db: SqliteDatabase, sessionId: string, options: ResolvedWriterLeaseOptions): WriterLease {
const now = Date.now();
const lease = acquireSessionLease(db, sessionId, uuidv7(), now, now + options.ttlMs);
const lease = acquireWriterLease(db, sessionId, uuidv7(), now, now + options.ttlMs);
if (!lease) throw activeWriterError(sessionId);
return lease;
}
@ -337,7 +337,7 @@ function requireSessionRow(db: SqliteDatabase, sessionId: string): SessionRow {
class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
private readonly db: SqliteDatabase;
private readonly metadata: SqliteSessionMetadata;
private readonly lease: SessionLease;
private readonly lease: WriterLease;
private readonly leaseOptions: ResolvedWriterLeaseOptions;
private readonly onRelease: () => void;
private readonly operations = new SerialOperationQueue();
@ -349,7 +349,7 @@ class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
constructor(
db: SqliteDatabase,
metadata: SqliteSessionMetadata,
lease: SessionLease,
lease: WriterLease,
leaseOptions: ResolvedWriterLeaseOptions,
onRelease: () => void,
) {
@ -371,7 +371,7 @@ class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
if (this.heartbeatTimer !== undefined) clearTimeout(this.heartbeatTimer);
try {
await this.operations.enqueue(() =>
this.db.transaction(() => releaseSessionLease(this.db, this.metadata.id, this.lease)),
this.db.transaction(() => releaseWriterLease(this.db, this.metadata.id, this.lease)),
);
} finally {
this.onRelease();
@ -385,7 +385,7 @@ class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
if (this.leaseError) throw this.leaseError;
return this.db.transaction(() => {
const now = Date.now();
if (!renewSessionLease(this.db, this.metadata.id, this.lease, now, now + this.leaseOptions.ttlMs)) {
if (!renewWriterLease(this.db, this.metadata.id, this.lease, now, now + this.leaseOptions.ttlMs)) {
this.leaseError = lostWriterError(this.metadata.id);
if (this.heartbeatTimer !== undefined) clearTimeout(this.heartbeatTimer);
throw this.leaseError;
@ -404,7 +404,7 @@ class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
if (this.closing || this.leaseError) return;
this.db.transaction(() => {
const now = Date.now();
if (!renewSessionLease(this.db, this.metadata.id, this.lease, now, now + this.leaseOptions.ttlMs)) {
if (!renewWriterLease(this.db, this.metadata.id, this.lease, now, now + this.leaseOptions.ttlMs)) {
this.leaseError = lostWriterError(this.metadata.id);
}
});
@ -419,7 +419,7 @@ class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
}
async getMetadata(): Promise<SqliteSessionMetadata> {
return structuredClone(this.metadata);
return decodeSessionMetadata(requireSessionRow(this.db, this.metadata.id), this.metadata.path);
}
isForSession(sessionId: string): boolean {
@ -628,24 +628,20 @@ function claimStorage(
): SqliteSessionStorage {
requireSessionRow(db, metadata.id);
const claimed = db.transaction(() => {
const lease = acquireWriterLease(db, metadata.id, leaseOptions);
const lease = claimWriterLease(db, metadata.id, leaseOptions);
const row = requireSessionRow(db, metadata.id);
readLanes(db, metadata.id);
return { lease, row };
});
return new SqliteSessionStorage(
db,
metadataFromRow(claimed.row, metadata.path),
decodeSessionMetadata(claimed.row, metadata.path),
claimed.lease,
leaseOptions,
onRelease,
);
}
function metadataFromRow(row: SessionRow, path: string): SqliteSessionMetadata {
return rowToMetadata(row, path);
}
export class SqliteSessionRepository
implements
SessionRepository<SqliteSessionMetadata, SqliteSessionCreateOptions, SqliteSessionListOptions>,
@ -673,7 +669,7 @@ export class SqliteSessionRepository
private sessionFromLease(
db: SqliteDatabase,
metadata: SqliteSessionMetadata,
lease: SessionLease,
lease: WriterLease,
): Session<SqliteSessionMetadata> {
let storage: SqliteSessionStorage;
storage = new SqliteSessionStorage(db, metadata, lease, this.leaseOptions, () => {
@ -715,10 +711,10 @@ export class SqliteSessionRepository
createSequence(db, id);
createStats(db, id);
createInitialLane(db, id);
return acquireWriterLease(db, id, this.leaseOptions);
return claimWriterLease(db, id, this.leaseOptions);
});
const row = requireSessionRow(db, id);
return this.sessionFromLease(db, metadataFromRow(row, path), lease);
return this.sessionFromLease(db, decodeSessionMetadata(row, path), lease);
});
}
@ -732,21 +728,22 @@ export class SqliteSessionRepository
await this.releaseStoragesForSession(metadata.id);
const db = await this.getDatabase();
db.transaction(() => {
const lease = acquireWriterLease(db, metadata.id, this.leaseOptions);
const lease = claimWriterLease(db, metadata.id, this.leaseOptions);
requireSessionRow(db, metadata.id);
rebuildBranchCache(db, metadata.id);
releaseSessionLease(db, metadata.id, lease);
releaseWriterLease(db, metadata.id, lease);
});
});
}
/** Reads the session catalog without acquiring or renewing per-session writer leases. */
async list(options: SqliteSessionListOptions = {}): Promise<SqliteSessionMetadata[]> {
return this.operations.enqueue(async () => {
const path = await this.getDatabasePath();
if (!resultOrThrow(await this.options.env.exists(path), `Failed to check database ${path}`)) return [];
const db = await this.getDatabase();
const rows = readSessionRows(db, options);
return rows.map((row) => metadataFromRow(row, path));
return rows.map((row) => decodeSessionMetadata(row, path));
});
}
@ -756,16 +753,16 @@ export class SqliteSessionRepository
const db = await this.getDatabase();
db.transaction(() => {
if (!sessionExists(db, metadata.id)) {
deleteSessionLease(db, metadata.id);
deleteWriterLease(db, metadata.id);
return;
}
acquireWriterLease(db, metadata.id, this.leaseOptions);
claimWriterLease(db, metadata.id, this.leaseOptions);
deleteBranchCache(db, metadata.id);
deleteFactRows(db, metadata.id);
deleteLaneRows(db, metadata.id);
deleteRecordRows(db, metadata.id);
deleteEntryRows(db, metadata.id);
deleteSessionLease(db, metadata.id);
deleteWriterLease(db, metadata.id);
deleteStats(db, metadata.id);
deleteSequence(db, metadata.id);
deleteSessionRow(db, metadata.id);
@ -780,7 +777,7 @@ export class SqliteSessionRepository
return this.operations.enqueue(async () => {
const db = await this.getDatabase();
const path = await this.getDatabasePath();
const sourceMetadata = metadataFromRow(requireSessionRow(db, source.id), path);
const sourceMetadata = decodeSessionMetadata(requireSessionRow(db, source.id), path);
const id = options.id ?? uuidv7();
if (sessionExists(db, id)) throw new SessionError("already_exists", `Session already exists: ${id}`);
@ -831,7 +828,7 @@ export class SqliteSessionRepository
);
const createdAt = Date.now();
const metadata = options.metadata ?? sourceMetadata.metadata;
let lease: SessionLease;
let lease: WriterLease;
try {
lease = db.transaction(() => {
@ -871,7 +868,7 @@ export class SqliteSessionRepository
setNextSequence(db, id, nextSeq);
for (const tip of branchTips) buildCachedBranch(db, id, tip);
return acquireWriterLease(db, id, this.leaseOptions);
return claimWriterLease(db, id, this.leaseOptions);
});
} catch (error) {
if (error instanceof SessionError) throw error;
@ -883,7 +880,7 @@ export class SqliteSessionRepository
}
const row = requireSessionRow(db, id);
return this.sessionFromLease(db, metadataFromRow(row, path), lease);
return this.sessionFromLease(db, decodeSessionMetadata(row, path), lease);
});
}

19
packages/session-backends/sqlite-node/src/sqlite/search-backend.ts
View file

@ -1,7 +1,7 @@
import type { SessionSearch, SessionSearchHit, SessionSearchOptions } from "@earendil-works/pi-agent-core";
import { getFileSystemResultOrThrow } from "@earendil-works/pi-agent-core";
import { applyMigrations } from "./migrations.ts";
import { rowToMetadata, type SessionRow } from "./storage/sessions.ts";
import { decodeSessionMetadata, type SessionRow } from "./storage/sessions.ts";
import type {
SqliteDatabase,
SqliteDatabaseFactory,
@ -104,7 +104,20 @@ class SqliteSessionSearch implements SessionSearch<SqliteSessionMetadata> {
const query = `"${text.replaceAll('"', '""')}"`;
const rows = db
.prepare(
"SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, se.id AS entry_id, se.timestamp, bm25(session_search_fts) AS score FROM session_search_fts JOIN entries se ON se.rowid = session_search_fts.rowid JOIN sessions s ON s.id = se.session_id WHERE session_search_fts MATCH ? AND (? IS NULL OR s.cwd = ?) ORDER BY score",
`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id,
(
SELECT f.value
FROM facts AS f
WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL
ORDER BY f.seq DESC
LIMIT 1
) AS session_name,
se.id AS entry_id, se.timestamp, bm25(session_search_fts) AS score
FROM session_search_fts
JOIN entries AS se ON se.rowid = session_search_fts.rowid
JOIN sessions AS s ON s.id = se.session_id
WHERE session_search_fts MATCH ? AND (? IS NULL OR s.cwd = ?)
ORDER BY score`,
)
.all<SessionRow & { entry_id: string; timestamp: string; score: number }>(
query,
@ -113,7 +126,7 @@ class SqliteSessionSearch implements SessionSearch<SqliteSessionMetadata> {
);
const path = await this.getDatabasePath();
return rows.map((row) => ({
metadata: rowToMetadata(row, path),
metadata: decodeSessionMetadata(row, path),
entryId: row.entry_id,
timestamp: row.timestamp,
score: row.score,

66
packages/session-backends/sqlite-node/src/sqlite/storage/sessions.ts
View file

@ -7,6 +7,7 @@ export interface SessionRow {
metadata: string | null;
cwd: string;
parent_session_id: string | null;
session_name: string | null;
}
export interface NewSessionRow {
@ -60,33 +61,74 @@ export function insertSessionRow(db: SqliteDatabase, session: NewSessionRow) {
export function readSessionRow(db: SqliteDatabase, sessionId: string) {
return db
.prepare("SELECT id, created_at, metadata, cwd, parent_session_id FROM sessions WHERE id = ?")
.prepare(
`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id,
(
SELECT f.value
FROM facts AS f
WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL
ORDER BY f.seq DESC
LIMIT 1
) AS session_name
FROM sessions AS s
WHERE s.id = ?`,
)
.get<SessionRow>(sessionId);
}
export function readSessionRows(db: SqliteDatabase, options: { cwd?: string } = {}) {
return options.cwd
? db
.prepare(
"SELECT id, created_at, metadata, cwd, parent_session_id FROM sessions WHERE cwd = ? ORDER BY created_at DESC",
)
.all<SessionRow>(options.cwd)
: db
.prepare("SELECT id, created_at, metadata, cwd, parent_session_id FROM sessions ORDER BY created_at DESC")
.all<SessionRow>();
const where = options.cwd === undefined ? "" : "WHERE s.cwd = ?";
return db
.prepare(
`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id,
(
SELECT f.value
FROM facts AS f
WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL
ORDER BY f.seq DESC
LIMIT 1
) AS session_name
FROM sessions AS s
${where}
ORDER BY s.created_at DESC`,
)
.all<SessionRow>(...(options.cwd === undefined ? [] : [options.cwd]));
}
export function deleteSessionRow(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM sessions WHERE id = ?").run(sessionId);
}
export function rowToMetadata(row: SessionRow, path: string): SqliteSessionMetadata {
function parseSessionName(value: string, sessionId: string): string {
let parsed: unknown;
try {
parsed = JSON.parse(value);
} catch (error) {
throw new SessionError(
"storage",
`Invalid SQLite session ${sessionId}: name is not valid JSON`,
error instanceof Error ? error : undefined,
);
}
if (typeof parsed !== "string") {
throw new SessionError("storage", `Invalid SQLite session ${sessionId}: name must be a string`);
}
return parsed;
}
export function decodeSessionMetadata(row: SessionRow, path: string): SqliteSessionMetadata {
const metadata = parseMetadata(row.metadata, row.id);
const name =
row.session_name === undefined || row.session_name === null
? undefined
: parseSessionName(row.session_name, row.id);
return {
id: row.id,
createdAt: Date.parse(row.created_at),
...(name === undefined ? {} : { name }),
cwd: row.cwd,
path,
parentSessionId: row.parent_session_id ?? undefined,
metadata: parseMetadata(row.metadata, row.id),
metadata,
};
}

28
packages/session-backends/sqlite-node/src/sqlite/storage/leases.ts → packages/session-backends/sqlite-node/src/sqlite/storage/writer-leases.ts
View file

@ -1,18 +1,18 @@
import type { SqliteDatabase } from "../types.ts";
export interface SessionLease {
export interface WriterLease {
ownerId: string;
fence: number;
expiresAtMs: number;
}
interface SessionLeaseRow {
interface WriterLeaseRow {
owner_id: string;
fence: number;
expires_at_ms: number;
}
export function acquireSessionLease(
export function acquireWriterLease(
db: SqliteDatabase,
sessionId: string,
ownerId: string,
@ -21,29 +21,29 @@ export function acquireSessionLease(
) {
const row = db
.prepare(
`INSERT INTO leases (session_id, owner_id, fence, expires_at_ms)
`INSERT INTO writer_leases (session_id, owner_id, fence, expires_at_ms)
VALUES (?, ?, 1, ?)
ON CONFLICT(session_id) DO UPDATE SET
owner_id = excluded.owner_id,
fence = leases.fence + 1,
fence = writer_leases.fence + 1,
expires_at_ms = excluded.expires_at_ms
WHERE leases.expires_at_ms <= ?
WHERE writer_leases.expires_at_ms <= ?
RETURNING owner_id, fence, expires_at_ms`,
)
.get<SessionLeaseRow>(sessionId, ownerId, expiresAtMs, now);
.get<WriterLeaseRow>(sessionId, ownerId, expiresAtMs, now);
return row === undefined ? undefined : { ownerId: row.owner_id, fence: row.fence, expiresAtMs: row.expires_at_ms };
}
export function renewSessionLease(
export function renewWriterLease(
db: SqliteDatabase,
sessionId: string,
lease: SessionLease,
lease: WriterLease,
now: number,
expiresAtMs: number,
) {
const result = db
.prepare(
`UPDATE leases
`UPDATE writer_leases
SET expires_at_ms = ?
WHERE session_id = ? AND owner_id = ? AND fence = ? AND expires_at_ms > ?`,
)
@ -52,14 +52,14 @@ export function renewSessionLease(
return result.changes === 1;
}
export function releaseSessionLease(db: SqliteDatabase, sessionId: string, lease: SessionLease) {
db.prepare("DELETE FROM leases WHERE session_id = ? AND owner_id = ? AND fence = ?").run(
export function releaseWriterLease(db: SqliteDatabase, sessionId: string, lease: WriterLease) {
db.prepare("DELETE FROM writer_leases WHERE session_id = ? AND owner_id = ? AND fence = ?").run(
sessionId,
lease.ownerId,
lease.fence,
);
}
export function deleteSessionLease(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM leases WHERE session_id = ?").run(sessionId);
export function deleteWriterLease(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM writer_leases WHERE session_id = ?").run(sessionId);
}

3
packages/session-backends/sqlite-node/src/sqlite/types.ts
View file

@ -32,6 +32,9 @@ export interface SqliteSessionMetadata extends SessionMetadata {
cwd: string;
path: string;
parentSessionId?: string;
/** Current session name projected from SQLite global facts. */
name?: string;
/** Opaque application-owned metadata. */
metadata?: Record<string, unknown>;
}

2
packages/session-backends/sqlite-node/test/migrations.test.ts
View file

@ -29,7 +29,7 @@ describe("SQLite migrations", () => {
"records",
"lane_moves",
"facts",
"leases",
"writer_leases",
]),
);
const sessionColumns = db.prepare("PRAGMA table_info(sessions)").all<{ name: string }>();

29
packages/session-backends/sqlite-node/test/repository.test.ts
View file

@ -257,6 +257,35 @@ END;
});
});
it.each([
["invalid JSON", "not json", "name is not valid JSON"],
["a non-string value", "{}", "name must be a string"],
])("rejects stored session names containing %s", async (_case, stored, message) => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
await using repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
await session.setName("valid name");
const db = await sqlite.open(databasePath);
try {
await db.prepare("UPDATE facts SET value = ? WHERE session_id = ? AND kind = 'name'").run(stored, "session-1");
} finally {
await db.close();
}
await expect(repo.list()).rejects.toMatchObject({
code: "storage",
message: expect.stringContaining(message),
});
await expect(session.getMetadata()).rejects.toMatchObject({
code: "storage",
message: expect.stringContaining(message),
});
});
it("fails loudly when a stored entry is read and cannot be decoded", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");

12
packages/session-backends/sqlite-node/test/search.test.ts
View file

@ -21,13 +21,21 @@ describe("SQLite FTS5 session search", () => {
const databasePath = join(root, "sessions.sqlite");
await using fixture = createSqliteFixture({ env, sqlite, databasePath });
const { repository: repo, search } = fixture;
const included = await repo.create({ cwd: root, id: "included" });
const included = await repo.create({ cwd: root, id: "included", metadata: { name: "application-owned" } });
const excluded = await repo.create({ cwd: `${root}/other`, id: "excluded" });
const entryId = await included.appendMessage(createUserMessage("Find the auth defect"));
await included.setName("Canonical name");
await excluded.appendMessage(createUserMessage("Find the auth defect"));
await expect(search.search({ text: "auth", cwd: root })).resolves.toEqual([
expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }),
expect.objectContaining({
entryId,
metadata: expect.objectContaining({
id: "included",
name: "Canonical name",
metadata: { name: "application-owned" },
}),
}),
]);
await expect(search.search({ text: "uth", cwd: root })).resolves.toEqual([
expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }),

67
packages/session-backends/sqlite-node/test/leases.test.ts → packages/session-backends/sqlite-node/test/writer-leases.test.ts
View file

@ -49,6 +49,63 @@ describe("SQLite session writer leases", () => {
).toThrow(message);
});
it("lists complete metadata without acquiring active sessions' writer leases", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
await using writerRepository = createRepository(root, databasePath);
await using readerRepository = createRepository(root, databasePath);
const first = await writerRepository.create({
cwd: root,
id: "session-1",
metadata: { profile: "reviewer" },
});
const second = await writerRepository.create({
cwd: root,
id: "session-2",
parentSessionId: "session-1",
metadata: { profile: "writer", name: "application-owned name" },
});
await first.setName("Review session");
await second.setName("Write session");
const [firstMetadata, secondMetadata] = await Promise.all([first.getMetadata(), second.getMetadata()]);
expect(firstMetadata).toMatchObject({ name: "Review session", metadata: { profile: "reviewer" } });
expect(secondMetadata).toMatchObject({
name: "Write session",
metadata: { profile: "writer", name: "application-owned name" },
});
const expected = [firstMetadata, secondMetadata];
const sqlite = createNodeSqliteFactory();
const inspection = await sqlite.open(databasePath);
let leasesBefore: { session_id: string; owner_id: string; fence: number; expires_at_ms: number }[];
try {
leasesBefore = await inspection
.prepare("SELECT session_id, owner_id, fence, expires_at_ms FROM writer_leases ORDER BY session_id")
.all();
} finally {
await inspection.close();
}
const listed = await readerRepository.list({ cwd: root });
expect([...listed].sort((left, right) => left.id.localeCompare(right.id))).toEqual(
[...expected].sort((left, right) => left.id.localeCompare(right.id)),
);
const afterList = await sqlite.open(databasePath);
try {
expect(
await afterList
.prepare("SELECT session_id, owner_id, fence, expires_at_ms FROM writer_leases ORDER BY session_id")
.all(),
).toEqual(leasesBefore);
} finally {
await afterList.close();
}
await expect(readerRepository.open(expected[0]!)).rejects.toMatchObject({
code: "storage",
message: expect.stringContaining("already has an active writer"),
});
});
it("rejects a second writer until the first session releases its claim", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
@ -78,7 +135,7 @@ describe("SQLite session writer leases", () => {
const sqlite = createNodeSqliteFactory();
const db = await sqlite.open(databasePath);
try {
await db.prepare("UPDATE leases SET expires_at_ms = 0 WHERE session_id = ?").run(metadata.id);
await db.prepare("UPDATE writer_leases SET expires_at_ms = 0 WHERE session_id = ?").run(metadata.id);
} finally {
await db.close();
}
@ -94,7 +151,7 @@ describe("SQLite session writer leases", () => {
let currentLease: { owner_id: string; fence: number } | undefined;
try {
currentLease = await inspection
.prepare("SELECT owner_id, fence FROM leases WHERE session_id = ?")
.prepare("SELECT owner_id, fence FROM writer_leases WHERE session_id = ?")
.get<{ owner_id: string; fence: number }>(metadata.id);
expect(currentLease?.fence).toBe(2);
} finally {
@ -105,7 +162,9 @@ describe("SQLite session writer leases", () => {
const afterStaleClose = await sqlite.open(databasePath);
try {
expect(
await afterStaleClose.prepare("SELECT owner_id, fence FROM leases WHERE session_id = ?").get(metadata.id),
await afterStaleClose
.prepare("SELECT owner_id, fence FROM writer_leases WHERE session_id = ?")
.get(metadata.id),
).toEqual(currentLease);
} finally {
await afterStaleClose.close();
@ -143,7 +202,7 @@ describe("SQLite session writer leases", () => {
try {
return (
await db
.prepare("SELECT expires_at_ms FROM leases WHERE session_id = ?")
.prepare("SELECT expires_at_ms FROM writer_leases WHERE session_id = ?")
.get<{ expires_at_ms: number }>(metadata.id)
)?.expires_at_ms;
} finally {