diff --git a/packages/core/src/session/event-log-compaction.ts b/packages/core/src/session/event-log-compaction.ts new file mode 100644 index 000000000000..370b6f03dc34 --- /dev/null +++ b/packages/core/src/session/event-log-compaction.ts @@ -0,0 +1,238 @@ +export * as SessionEventLogCompaction from "./event-log-compaction" + +import { isDeepStrictEqual } from "node:util" +import { sql } from "drizzle-orm" +import { Effect } from "effect" +import { Event } from "@opencode-ai/schema/event" +import type { Database } from "../database/database" + +const DEFAULT_LIMIT = 1000 +const MAX_LIMIT = 10_000 +const checkpointType = Event.versionedType(Event.Compacted.type, Event.Compacted.durable!.version) + +type Policy = (typeof policies)[number] +type Candidate = { + id: string + aggregateID: string + type: string + bytes: number + supersededBy: string + latestData: string + projectionData: string + workspaceID: string | null + ownerID: string | null +} + +export type Options = { + readonly aggregateID?: string + readonly all?: boolean + readonly apply?: boolean + readonly limit?: number +} + +export type Report = { + readonly dryRun: boolean + readonly candidates: number + readonly rewritten: number + readonly projectionMismatches: number + readonly compatibilityRejected: number + readonly malformed: number + readonly payloadBytesReclaimed: number + readonly hasMore: boolean + readonly continuation: string + readonly byType: Readonly> +} + +export type Status = { + readonly events: number + readonly payloadBytes: number + readonly compactableEvents: number + readonly recommended: boolean +} + +const policies = [ + { type: "message.updated.1", path: "$.info.id", table: "message", fields: ["id", "sessionID"] }, + { type: "message.part.updated.1", path: "$.part.id", table: "part", fields: ["id", "messageID", "sessionID"] }, +] as const + +function validate(options: Options) { + if (Boolean(options.aggregateID) === Boolean(options.all)) { + throw new Error("Specify exactly one compaction scope: aggregateID or all") + } + const limit = options.limit ?? DEFAULT_LIMIT + if (!Number.isSafeInteger(limit) || limit <= 0 || limit > MAX_LIMIT) { + throw new Error(`limit must be a positive integer no greater than ${MAX_LIMIT}`) + } + return limit +} + +function continuation(options: Options, limit: number) { + const scope = options.all ? "--all" : `--session ${options.aggregateID}` + return `opencode db compact-events ${scope} --apply --limit ${limit}` +} + +function checkpoint(candidate: Candidate) { + return { aggregateID: candidate.aggregateID, supersededType: candidate.type, supersededBy: candidate.supersededBy } +} + +function reclaimedBytes(candidate: Candidate) { + return Math.max(0, candidate.bytes - Buffer.byteLength(JSON.stringify(checkpoint(candidate)))) +} + +function projectedData(candidate: Candidate, policy: Policy) { + const latest = JSON.parse(candidate.latestData) as { info?: Record; part?: Record } + const value = policy.type === "message.updated.1" ? latest.info : latest.part + if (!value) return undefined + const result = { ...value } + for (const field of policy.fields) delete result[field] + return result +} + +function safe(candidate: Candidate, policy: Policy) { + try { + return isDeepStrictEqual(projectedData(candidate, policy), JSON.parse(candidate.projectionData)) + } catch { + return false + } +} + +const candidates = Effect.fn("SessionEventLogCompaction.candidates")(function* ( + db: Database.Interface["db"], + options: Options, + limit: number, +) { + const rows = new Array() + for (const policy of policies) { + if (rows.length >= limit + 1) break + const selected = yield* db + .all( + sql` + WITH latest AS ( + SELECT aggregate_id, json_extract(data, ${policy.path}) AS entity_id, MAX(seq) AS seq + FROM event + WHERE type = ${policy.type} + AND json_valid(data) + ${options.all ? sql`` : sql`AND aggregate_id = ${options.aggregateID!}`} + GROUP BY aggregate_id, json_extract(data, ${policy.path}) + ) + SELECT event.id, event.aggregate_id AS aggregateID, event.type, length(event.data) AS bytes, + replacement.id AS supersededBy, replacement.data AS latestData, + projection.data AS projectionData, session.workspace_id AS workspaceID, + event_sequence.owner_id AS ownerID + FROM event + JOIN latest ON latest.aggregate_id = event.aggregate_id + AND latest.entity_id = json_extract(CASE WHEN json_valid(event.data) THEN event.data END, ${policy.path}) + JOIN event AS replacement ON replacement.aggregate_id = latest.aggregate_id + AND replacement.type = ${policy.type} AND replacement.seq = latest.seq + JOIN ${sql.identifier(policy.table)} AS projection ON projection.id = latest.entity_id + AND projection.session_id = latest.aggregate_id + JOIN session ON session.id = event.aggregate_id + JOIN event_sequence ON event_sequence.aggregate_id = event.aggregate_id + WHERE event.type = ${policy.type} AND json_valid(event.data) AND event.seq < latest.seq + ORDER BY event.aggregate_id, event.seq + LIMIT ${limit + 1 - rows.length} + `, + ) + .pipe(Effect.orDie) + rows.push(...selected.map((candidate) => ({ ...candidate, policy }))) + } + return rows +}) + +const malformed = Effect.fn("SessionEventLogCompaction.malformed")(function* ( + db: Database.Interface["db"], + options: Options, +) { + const row = yield* db + .get<{ count: number }>( + sql` + SELECT count(*) AS count FROM event + WHERE type IN (${policies[0].type}, ${policies[1].type}) AND NOT json_valid(data) + ${options.all ? sql`` : sql`AND aggregate_id = ${options.aggregateID!}`} + `, + ) + .pipe(Effect.orDie) + return row?.count ?? 0 +}) + +function report( + rows: ReadonlyArray, + malformedCount: number, + options: Options, + limit: number, +) { + const inspected = rows.slice(0, limit) + const safeRows = inspected.filter((candidate) => safe(candidate, candidate.policy)) + const eligible = safeRows.filter((candidate) => candidate.workspaceID === null && candidate.ownerID === null) + const byType: Record = {} + for (const candidate of eligible) { + const summary = byType[candidate.type] ?? { events: 0, payloadBytesReclaimed: 0 } + summary.events++ + summary.payloadBytesReclaimed += reclaimedBytes(candidate) + byType[candidate.type] = summary + } + return { + dryRun: !options.apply, + candidates: eligible.length, + rewritten: options.apply ? eligible.length : 0, + projectionMismatches: inspected.filter((candidate) => !safe(candidate, candidate.policy)).length, + compatibilityRejected: safeRows.length - eligible.length, + malformed: malformedCount, + payloadBytesReclaimed: eligible.reduce((total, candidate) => total + reclaimedBytes(candidate), 0), + hasMore: rows.length > limit, + continuation: continuation(options, limit), + byType, + } satisfies Report +} + +/** + * Replaces only locally-owned, projection-verified snapshots. Sync has no + * version negotiation for checkpoint markers, so aggregates with a workspace + * or sync owner are always dry-run-only until that protocol exists. + */ +export const compact = Effect.fn("SessionEventLogCompaction.compact")(function* ( + db: Database.Interface["db"], + options: Options, +) { + const limit = validate(options) + if (!options.apply) + return report(yield* candidates(db, options, limit), yield* malformed(db, options), options, limit) + return yield* db + .transaction( + () => + Effect.gen(function* () { + const selected = yield* candidates(db, options, limit) + const result = report(selected, yield* malformed(db, options), options, limit) + const eligible = selected + .slice(0, limit) + .filter( + (candidate) => + safe(candidate, candidate.policy) && candidate.workspaceID === null && candidate.ownerID === null, + ) + yield* Effect.forEach(eligible, (candidate) => + db.run(sql` + UPDATE event SET type = ${checkpointType}, data = ${JSON.stringify(checkpoint(candidate))} + WHERE id = ${candidate.id} AND type = ${candidate.type} + `), + ) + return result + }), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) +}) + +export const status = Effect.fn("SessionEventLogCompaction.status")(function* (db: Database.Interface["db"]) { + const row = yield* db + .get<{ events: number; payloadBytes: number; compactableEvents: number }>( + sql` + SELECT count(*) AS events, + coalesce(sum(length(data)), 0) AS payloadBytes, + coalesce(sum(type IN (${policies[0].type}, ${policies[1].type})), 0) AS compactableEvents + FROM event + `, + ) + .pipe(Effect.orDie) + const result = row ?? { events: 0, payloadBytes: 0, compactableEvents: 0 } + return { ...result, recommended: result.compactableEvents > DEFAULT_LIMIT } satisfies Status +}) diff --git a/packages/core/test/session-event-log-compaction.test.ts b/packages/core/test/session-event-log-compaction.test.ts new file mode 100644 index 000000000000..24a99c2c904a --- /dev/null +++ b/packages/core/test/session-event-log-compaction.test.ts @@ -0,0 +1,156 @@ +import { describe, expect } from "bun:test" +import { Effect } from "effect" +import { asc, eq, sql } from "drizzle-orm" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { SessionEventLogCompaction } from "@opencode-ai/core/session/event-log-compaction" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionV1 } from "@opencode-ai/schema/session-v1" +import { SessionID } from "@opencode-ai/schema/session-id" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) + +describe("SessionEventLogCompaction", () => { + it.effect("keeps replay and the message projection identical while reclaiming superseded snapshots", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + const events = yield* EventV2.Service + const sessionID = SessionID.descending("ses_event_log_compaction") + const messageID = SessionV1.MessageID.ascending("msg_event_log_compaction") + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "compaction", + directory: "/project", + title: "compaction", + version: "test", + }) + .run() + const message = (agent: string) => ({ + id: messageID, + sessionID, + role: "user" as const, + time: { created: 1 }, + agent, + model: { providerID: ProviderV2.ID.make("provider"), modelID: ModelV2.ID.make("model") }, + }) + + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("after") }) + const projection = yield* db.select().from(MessageTable).where(eq(MessageTable.id, messageID)).get() + expect(yield* SessionEventLogCompaction.status(db)).toMatchObject({ events: 2, compactableEvents: 2 }) + const dryRun = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID }) + + expect(dryRun).toMatchObject({ candidates: 1, rewritten: 0, payloadBytesReclaimed: expect.any(Number) }) + expect(projection?.data).toMatchObject({ agent: "after" }) + + const applied = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + const compacted = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .orderBy(asc(EventTable.seq)) + .all() + + expect(applied).toMatchObject({ candidates: 1, rewritten: 1 }) + expect(compacted.map((event) => event.seq)).toEqual([0, 1]) + expect(compacted.map((event) => event.type)).toEqual(["event.compacted.1", "message.updated.1"]) + expect(yield* db.select().from(MessageTable).where(eq(MessageTable.id, messageID)).get()).toEqual(projection) + + expect( + yield* events.replayAll( + compacted.map((event) => ({ + id: event.id, + aggregateID: event.aggregate_id, + seq: event.seq, + type: event.type, + data: event.data, + })), + ), + ).toBe(sessionID) + + const partID = SessionV1.PartID.ascending("prt_event_log_compaction") + const part = (text: string) => ({ id: partID, sessionID, messageID, type: "text" as const, text }) + yield* events.publish(SessionV1.Event.PartUpdated, { sessionID, part: part("one"), time: 1 }) + yield* events.publish(SessionV1.Event.PartUpdated, { sessionID, part: part("two"), time: 2 }) + yield* events.publish(SessionV1.Event.PartUpdated, { sessionID, part: part("three"), time: 3 }) + + const firstPartBatch = yield* SessionEventLogCompaction.compact(db, { + aggregateID: sessionID, + apply: true, + limit: 1, + }) + const secondPartBatch = yield* SessionEventLogCompaction.compact(db, { + aggregateID: sessionID, + apply: true, + limit: 1, + }) + const idempotent = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + expect(firstPartBatch).toMatchObject({ candidates: 1, rewritten: 1, hasMore: true }) + expect(firstPartBatch.continuation).toBe(`opencode db compact-events --session ${sessionID} --apply --limit 1`) + expect(secondPartBatch).toMatchObject({ candidates: 1, rewritten: 1 }) + expect(idempotent).toMatchObject({ candidates: 0, rewritten: 0 }) + + const mismatchID = SessionV1.MessageID.ascending("msg_event_log_mismatch") + const mismatch = (agent: string) => ({ ...message(agent), id: mismatchID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: mismatch("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: mismatch("after") }) + yield* db.run(sql`UPDATE message SET data = ${JSON.stringify({ agent: "stale" })} WHERE id = ${mismatchID}`) + expect(yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).toMatchObject({ + projectionMismatches: 1, + }) + + const malformedID = SessionV1.MessageID.ascending("msg_event_log_malformed") + const malformed = (agent: string) => ({ ...message(agent), id: malformedID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: malformed("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: malformed("after") }) + yield* db.run(sql`UPDATE event SET data = 'not json' WHERE type = 'message.updated.1' AND seq = 6`) + expect(yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).toMatchObject({ malformed: 1 }) + + const appendedID = SessionV1.MessageID.ascending("msg_event_log_appended") + const appended = (agent: string) => ({ ...message(agent), id: appendedID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: appended("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: appended("middle") }) + expect((yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).candidates).toBeGreaterThan(0) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: appended("latest") }) + yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + expect((yield* db.select().from(MessageTable).where(eq(MessageTable.id, appendedID)).get())?.data).toMatchObject({ + agent: "latest", + }) + + const ownedID = SessionV1.MessageID.ascending("msg_event_log_owned") + const owned = (agent: string) => ({ ...message(agent), id: ownedID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: owned("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: owned("after") }) + yield* events.claim(sessionID, "sync-owner") + const rejected = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + expect(rejected).toMatchObject({ candidates: 0, rewritten: 0, compatibilityRejected: expect.any(Number) }) + + for (const options of [ + {}, + { aggregateID: sessionID, all: true }, + { aggregateID: sessionID, limit: 0 }, + { aggregateID: sessionID, limit: 10_001 }, + ]) { + const exit = yield* SessionEventLogCompaction.compact(db, options).pipe(Effect.exit) + expect(String(exit)).toContain("Error") + } + }), + ) +}) diff --git a/packages/opencode/src/cli/cmd/db.ts b/packages/opencode/src/cli/cmd/db.ts index 9e7e37e18e91..ec5acd801a43 100644 --- a/packages/opencode/src/cli/cmd/db.ts +++ b/packages/opencode/src/cli/cmd/db.ts @@ -1,9 +1,10 @@ import type { Argv } from "yargs" import { spawn } from "child_process" import { Database } from "@opencode-ai/core/database/database" +import { SessionEventLogCompaction } from "@opencode-ai/core/session/event-log-compaction" import { Effect } from "effect" import { sql } from "drizzle-orm" -import { effectCmd } from "../effect-cmd" +import { effectCmd, fail } from "../effect-cmd" const QueryCommand = effectCmd({ command: "$0 [query]", @@ -51,12 +52,54 @@ const PathCommand = effectCmd({ }), }) +const CompactEventsCommand = effectCmd({ + command: "compact-events", + describe: "replace superseded message and part snapshots with replay-safe checkpoints", + instance: false, + builder: (yargs: Argv) => + yargs + .option("apply", { type: "boolean", default: false, describe: "write checkpoints; default is dry-run" }) + .option("session", { type: "string", describe: "compact one session aggregate" }) + .option("all", { type: "boolean", default: false, describe: "inspect all session aggregates" }) + .option("limit", { type: "number", describe: "maximum snapshots per bounded batch" }), + handler: Effect.fn("Cli.db.compactEvents")(function* (args: { + apply: boolean + session?: string + all: boolean + limit?: number + }) { + const { db } = yield* Database.Service + const report = yield* SessionEventLogCompaction.compact(db, { + aggregateID: args.session, + all: args.all, + apply: args.apply, + limit: args.limit, + }).pipe(Effect.catchDefect((error) => fail(error instanceof Error ? error.message : String(error)))) + console.log(JSON.stringify(report, null, 2)) + }), +}) + +const EventLogStatusCommand = effectCmd({ + command: "event-log-status", + describe: "report event-log growth and compaction recommendation", + instance: false, + handler: Effect.fn("Cli.db.eventLogStatus")(function* () { + const { db } = yield* Database.Service + console.log(JSON.stringify(yield* SessionEventLogCompaction.status(db), null, 2)) + }), +}) + export const DbCommand = effectCmd({ command: "db", describe: "database tools", instance: false, builder: (yargs: Argv) => { - return yargs.command(QueryCommand).command(PathCommand).demandCommand() + return yargs + .command(QueryCommand) + .command(PathCommand) + .command(CompactEventsCommand) + .command(EventLogStatusCommand) + .demandCommand() }, handler: Effect.fn("Cli.db")(function* () {}), }) diff --git a/packages/schema/src/durable-event-manifest.ts b/packages/schema/src/durable-event-manifest.ts index acdcb3e9d568..a6f0822a3db3 100644 --- a/packages/schema/src/durable-event-manifest.ts +++ b/packages/schema/src/durable-event-manifest.ts @@ -10,6 +10,7 @@ export const SessionDurable = { } as const export const Durable = Event.durable([ + Event.Compacted, ...SessionV1.Event.Definitions.filter((definition) => definition.durable !== undefined), ...SessionEvent.DurableDefinitions, ]) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index 0d6ec9775aa4..94078269c64c 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -12,6 +12,21 @@ export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe( ) export type ID = typeof ID.Type +/** + * Replaces a durable snapshot that is represented by a later full snapshot. + * Keeping this small marker preserves the aggregate's contiguous sequence for + * strict sync replay while allowing the superseded payload to be reclaimed. + */ +export const Compacted = define({ + type: "event.compacted", + durable: { version: 1, aggregate: "aggregateID" }, + schema: { + aggregateID: Schema.String, + supersededType: Schema.String, + supersededBy: Schema.String, + }, +}) + export type Definition< Type extends string = string, DataSchema extends Schema.Codec = Schema.Codec,