mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(server): maintain native event cursors transactionally
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
83b5f7d486
commit
31479d4f6c
21 files changed
+1512
-61
No files matched your search
+16
-5
@@ -87,11 +87,22 @@ still locks the task before the run and reconciles pending steering from durable
|
||||
acknowledgements on every request. Default identity reads retain the full run.
|
||||
Stored payloads and separate native tool authorization checks remain unchanged.
|
||||
|
||||
Native event acknowledgements read a count and maximum through the existing
|
||||
source-sequence index. A stream with gaps uses indexed lookups for its first
|
||||
missing successor. The acknowledgement keeps its contiguous-prefix rule without
|
||||
returning every historical sequence number. All events stay stored. Counting the
|
||||
index entries still grows with the stream length.
|
||||
Native event acknowledgements use company-scoped `heartbeat_run_event_cursors`
|
||||
and disjoint `heartbeat_run_event_source_ranges` derived from retained events.
|
||||
The run lock protects sequence allocation, source range merges and event insertion
|
||||
in one transaction. In-order ingestion does not scan historical source sequences;
|
||||
closing a gap joins range metadata rather than loading its event suffix.
|
||||
|
||||
The run's `event_cursor_projection_seq` fences this optimization against older
|
||||
writers. A stale marker uses the correct retained-event calculation and sets
|
||||
`event_cursor_rebuild_pending` through a transaction-aware work notification.
|
||||
The delivery coordinator reconstructs explicit pending work at startup and
|
||||
rebuilds at most 512 event metadata rows per transaction. It releases the run
|
||||
lock between pages and stops when the marker catches up. Partial projections
|
||||
never supply acknowledgements. New runs use the fast path immediately; existing
|
||||
histories keep their previous append cost until rebuilding completes. The additive
|
||||
migration does not scan or rewrite event history. Deleting a run cascades its
|
||||
derived source state.
|
||||
|
||||
Conversation recovery reads only the saved run fields used to settle a turn.
|
||||
Settlement locks and reads current conversation state without loading the task
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
# Native event ingestion: implementation and acceptance
|
||||
|
||||
Goal: reduce durable native ingestion work and improve visible streaming under
|
||||
load, with no intentional collection delay. Deliver a cursor PR followed by a
|
||||
dependent batching PR, each with correctness and measured incremental evidence.
|
||||
|
||||
## Fixed workload and measurement contract
|
||||
|
||||
- B0 is `2e37c73c05fe84aead1883b72fdf6b1653b58c2a`; B1 contains only the cursor
|
||||
change; B2 adds batching. Record exact tested commits and dirty-file hashes.
|
||||
- Use the actual append service with disposable PostgreSQL and 1k, 10k and 100k
|
||||
histories. Include singleton writes, duplicate replay, sparse sequences and
|
||||
late gap closure. Setup and warmup are outside measured intervals.
|
||||
- Qualify the harness with a small case. Measure baseline against itself before
|
||||
candidate comparison; retain the observed repeatability envelope. Compare at
|
||||
least five alternating pairs. Report median, spread, samples and all failures.
|
||||
- Burst fixtures emit 16, 128 and 512 events, with 1, 10 and 100 concurrent runs.
|
||||
Include final drainage and independently compare stored identities/payloads.
|
||||
- Application proof requires a production build, a separate arrival/load clock,
|
||||
and browser render observations. Measure SQL/transactions, lock waits, CPU,
|
||||
backlog/memory, event-loop delay and emission-to-render latency. A service
|
||||
benchmark alone cannot establish browser or deployed improvement.
|
||||
|
||||
## Acceptance gates
|
||||
|
||||
| Outcome | Gate |
|
||||
| --- | --- |
|
||||
| Flat steady-state cursor work | Indexed cursor/range operations do not scan retained history; long-history append cost improves beyond baseline noise. |
|
||||
| Useful batching | Burst completion rate improves beyond noise in at least four of five pairs; fewer transactions per event; no missing work or hidden backlog. |
|
||||
| Immediate streaming | First and isolated text start persistence immediately; no timer or minimum batch size; sparse render latency does not regress beyond baseline noise. |
|
||||
| Loaded user experience | Combined change improves browser delivery latency beyond baseline noise under the same offered load. |
|
||||
| Correctness and controls | Replay identities, gaps, conflicts, company scope, outer rollback and commit-before-effect barriers remain correct. Stop and control latency do not regress. |
|
||||
| Compatibility and quiet | Stale projections use the correct fallback; bounded rebuild survives restart/old writers and stops all recurring SQL after draining. |
|
||||
|
||||
Do not lower these gates after measuring a candidate. An inconclusive or failing
|
||||
comparison needs diagnosis, not a successful-performance claim.
|
||||
|
||||
## Implementation ownership and boundaries
|
||||
|
||||
- Cursor/range state belongs with heartbeat event persistence. It remains a
|
||||
derived optimization, fenced by the run's committed global-sequence marker.
|
||||
Existing histories use the correct aggregate fallback while a concrete rebuild
|
||||
obligation runs in 512-row transactions through the existing work coordinator.
|
||||
- The native session runtime owns bounded read-ahead through its existing
|
||||
transformed iterator: 64 events / 64 KiB, one pending read and one append.
|
||||
Start the first append immediately; submit ready text as soon as the prior
|
||||
write settles. Nontext/P0 events remain ordering barriers.
|
||||
- The control-plane port owns validation, ordered identity redaction and outer
|
||||
revocation fencing. Database batches commit before receipts, publication and
|
||||
callbacks. No numbered events are merged.
|
||||
- Raw transport ACK/recovery policy is outside this change. Database receipts
|
||||
remain durable; raw ordinary-output ACKs do not universally promise transcript
|
||||
durability. The upstream normalized producer queue remains a separate limit.
|
||||
|
||||
## Verification record
|
||||
|
||||
Baseline SQL-only investigation: PostgreSQL 18.1, Node 26.4.0, five warm samples
|
||||
per size. The current count/max query examined 1k / 10k / 100k index entries, with
|
||||
median execution times 0.063 / 0.504 / 5.243 ms. This is mechanism evidence only.
|
||||
|
||||
Current state: implementation in progress. No candidate speedup or browser result
|
||||
has been established. Dependencies are installed and the additive migration was
|
||||
generated with the repository command. The lead owns integration and measurement
|
||||
windows; workers own cursor services, benchmark harness and isolated runtime work.
|
||||
|
||||
Before delivery: run focused PostgreSQL/runtime tests, migration checks, required
|
||||
typecheck/test/build gates, then the real browser native-run/tool/result/reload
|
||||
and Stop journeys. Record target, exact revision, provider versus fixture, run
|
||||
IDs, persisted outcomes and accessible evidence. Each PR uses the repository
|
||||
template and includes its own measurements and remaining limitations. Deployed
|
||||
claims require the exact serving revision and a separate matched verification.
|
||||
@@ -184,8 +184,13 @@ readers of the whole payload must apply `decodeRunEventPayload`. The column
|
||||
remains JSONB and requires no schema migration.
|
||||
|
||||
The writer locks the native `heartbeat_runs` row and allocates the existing
|
||||
per-run `seq` cursor. A byte-equivalent retry reuses the first row; a changed
|
||||
retry or source-sequence gap is rejected. Company, issue, agent, run, session,
|
||||
per-run `seq` cursor. A byte-equivalent retry with both matching source identities
|
||||
reuses the first row; changed identities or payloads fail closed. Shared ingestion
|
||||
can retain out-of-order source events but acknowledges only the contiguous prefix.
|
||||
The strict PRP coordinator rejects new source gaps. Transactional source cursors
|
||||
and received-range metadata avoid repeated history scans. A stale projection
|
||||
uses the retained-event calculation while bounded rebuild work catches up; it
|
||||
cannot over-acknowledge from partial state. Company, issue, agent, run, session,
|
||||
and runner-source bindings must match the persisted native run. Bootstrap
|
||||
tickets, reconnect leases, authentication proofs, encryption keys, and raw
|
||||
credential material are never written to the run log.
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
CREATE TABLE "heartbeat_run_event_cursors" (
|
||||
"company_id" uuid NOT NULL,
|
||||
"run_id" uuid NOT NULL,
|
||||
"source_instance_id" text NOT NULL,
|
||||
"contiguous_source_seq" bigint DEFAULT 0 NOT NULL,
|
||||
"pending_range_count" integer DEFAULT 0 NOT NULL,
|
||||
CONSTRAINT "heartbeat_run_event_cursors_company_id_run_id_source_instance_id_pk" PRIMARY KEY("company_id","run_id","source_instance_id"),
|
||||
CONSTRAINT "heartbeat_run_event_cursors_nonnegative" CHECK ("heartbeat_run_event_cursors"."contiguous_source_seq" >= 0 and "heartbeat_run_event_cursors"."pending_range_count" >= 0)
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE TABLE "heartbeat_run_event_source_ranges" (
|
||||
"company_id" uuid NOT NULL,
|
||||
"run_id" uuid NOT NULL,
|
||||
"source_instance_id" text NOT NULL,
|
||||
"start_seq" bigint NOT NULL,
|
||||
"end_seq" bigint NOT NULL,
|
||||
CONSTRAINT "heartbeat_run_event_source_ranges_company_id_run_id_source_instance_id_start_seq_pk" PRIMARY KEY("company_id","run_id","source_instance_id","start_seq"),
|
||||
CONSTRAINT "heartbeat_run_event_source_ranges_positive" CHECK ("heartbeat_run_event_source_ranges"."start_seq" >= 1 and "heartbeat_run_event_source_ranges"."end_seq" >= "heartbeat_run_event_source_ranges"."start_seq")
|
||||
);
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_runs" ADD COLUMN "event_cursor_projection_seq" bigint DEFAULT 0 NOT NULL;--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_runs" ADD COLUMN "event_cursor_rebuild_pending" boolean DEFAULT false NOT NULL;--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_run_event_cursors" ADD CONSTRAINT "heartbeat_run_event_cursors_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_run_event_cursors" ADD CONSTRAINT "heartbeat_run_event_cursors_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_run_event_source_ranges" ADD CONSTRAINT "heartbeat_run_event_source_ranges_cursor_fk" FOREIGN KEY ("company_id","run_id","source_instance_id") REFERENCES "public"."heartbeat_run_event_cursors"("company_id","run_id","source_instance_id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
CREATE INDEX "heartbeat_runs_event_cursor_rebuild_idx" ON "heartbeat_runs" USING btree ("company_id","id") WHERE "heartbeat_runs"."event_cursor_rebuild_pending" = true;
|
||||
+515
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"id": "8f75a1bb-32d9-4aed-9f93-134a6ae71f79",
|
||||
"prevId": "52cf22f3-3bd9-4b0f-8d23-b422bb0d2540",
|
||||
"id": "ecddf6e1-b7e6-4b93-aecf-5238948438ee",
|
||||
"prevId": "8d9f4abf-57c7-443a-aa8b-3e3fd4f09dae",
|
||||
"version": "7",
|
||||
"dialect": "postgresql",
|
||||
"tables": {
|
||||
@@ -2233,6 +2233,74 @@
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.agent_file_roots": {
|
||||
"name": "agent_file_roots",
|
||||
"schema": "",
|
||||
"columns": {
|
||||
"company_id": {
|
||||
"name": "company_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"agent_id": {
|
||||
"name": "agent_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"generation": {
|
||||
"name": "generation",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"entries": {
|
||||
"name": "entries",
|
||||
"type": "jsonb",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"updated_at": {
|
||||
"name": "updated_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "now()"
|
||||
}
|
||||
},
|
||||
"indexes": {},
|
||||
"foreignKeys": {
|
||||
"agent_file_roots_company_id_agent_id_agents_company_id_id_fk": {
|
||||
"name": "agent_file_roots_company_id_agent_id_agents_company_id_id_fk",
|
||||
"tableFrom": "agent_file_roots",
|
||||
"tableTo": "agents",
|
||||
"columnsFrom": [
|
||||
"company_id",
|
||||
"agent_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"company_id",
|
||||
"id"
|
||||
],
|
||||
"onDelete": "cascade",
|
||||
"onUpdate": "no action"
|
||||
}
|
||||
},
|
||||
"compositePrimaryKeys": {
|
||||
"agent_file_roots_company_id_agent_id_pk": {
|
||||
"name": "agent_file_roots_company_id_agent_id_pk",
|
||||
"columns": [
|
||||
"company_id",
|
||||
"agent_id"
|
||||
]
|
||||
}
|
||||
},
|
||||
"uniqueConstraints": {},
|
||||
"policies": {},
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.agent_identity_keys": {
|
||||
"name": "agent_identity_keys",
|
||||
"schema": "",
|
||||
@@ -29705,6 +29773,241 @@
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.execution_workspace_snapshots": {
|
||||
"name": "execution_workspace_snapshots",
|
||||
"schema": "",
|
||||
"columns": {
|
||||
"id": {
|
||||
"name": "id",
|
||||
"type": "uuid",
|
||||
"primaryKey": true,
|
||||
"notNull": true,
|
||||
"default": "gen_random_uuid()"
|
||||
},
|
||||
"company_id": {
|
||||
"name": "company_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"execution_workspace_id": {
|
||||
"name": "execution_workspace_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"source_operation_id": {
|
||||
"name": "source_operation_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"source_run_id": {
|
||||
"name": "source_run_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"parent_snapshot_id": {
|
||||
"name": "parent_snapshot_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"generation": {
|
||||
"name": "generation",
|
||||
"type": "integer",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"epoch": {
|
||||
"name": "epoch",
|
||||
"type": "integer",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"status": {
|
||||
"name": "status",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "'pending'"
|
||||
},
|
||||
"format": {
|
||||
"name": "format",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "'paperclip.remote-workspace-archive.v1'"
|
||||
},
|
||||
"object_key": {
|
||||
"name": "object_key",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"sha256": {
|
||||
"name": "sha256",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"byte_size": {
|
||||
"name": "byte_size",
|
||||
"type": "bigint",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"key": {
|
||||
"name": "key",
|
||||
"type": "jsonb",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"manifest_digest": {
|
||||
"name": "manifest_digest",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"metadata": {
|
||||
"name": "metadata",
|
||||
"type": "jsonb",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"delete_after": {
|
||||
"name": "delete_after",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"last_error": {
|
||||
"name": "last_error",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"created_at": {
|
||||
"name": "created_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "now()"
|
||||
},
|
||||
"updated_at": {
|
||||
"name": "updated_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "now()"
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"execution_workspace_snapshots_operation_uniq": {
|
||||
"name": "execution_workspace_snapshots_operation_uniq",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "execution_workspace_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "source_operation_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": true,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"execution_workspace_snapshots_generation_uniq": {
|
||||
"name": "execution_workspace_snapshots_generation_uniq",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "execution_workspace_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "generation",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": true,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"execution_workspace_snapshots_workspace_idx": {
|
||||
"name": "execution_workspace_snapshots_workspace_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "execution_workspace_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"execution_workspace_snapshots_cleanup_idx": {
|
||||
"name": "execution_workspace_snapshots_cleanup_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "status",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "delete_after",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
}
|
||||
},
|
||||
"foreignKeys": {},
|
||||
"compositePrimaryKeys": {},
|
||||
"uniqueConstraints": {},
|
||||
"policies": {},
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.execution_workspaces": {
|
||||
"name": "execution_workspaces",
|
||||
"schema": "",
|
||||
@@ -33155,6 +33458,168 @@
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.heartbeat_run_event_cursors": {
|
||||
"name": "heartbeat_run_event_cursors",
|
||||
"schema": "",
|
||||
"columns": {
|
||||
"company_id": {
|
||||
"name": "company_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"run_id": {
|
||||
"name": "run_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"source_instance_id": {
|
||||
"name": "source_instance_id",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"contiguous_source_seq": {
|
||||
"name": "contiguous_source_seq",
|
||||
"type": "bigint",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": 0
|
||||
},
|
||||
"pending_range_count": {
|
||||
"name": "pending_range_count",
|
||||
"type": "integer",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": 0
|
||||
}
|
||||
},
|
||||
"indexes": {},
|
||||
"foreignKeys": {
|
||||
"heartbeat_run_event_cursors_company_id_companies_id_fk": {
|
||||
"name": "heartbeat_run_event_cursors_company_id_companies_id_fk",
|
||||
"tableFrom": "heartbeat_run_event_cursors",
|
||||
"tableTo": "companies",
|
||||
"columnsFrom": [
|
||||
"company_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"id"
|
||||
],
|
||||
"onDelete": "no action",
|
||||
"onUpdate": "no action"
|
||||
},
|
||||
"heartbeat_run_event_cursors_run_id_heartbeat_runs_id_fk": {
|
||||
"name": "heartbeat_run_event_cursors_run_id_heartbeat_runs_id_fk",
|
||||
"tableFrom": "heartbeat_run_event_cursors",
|
||||
"tableTo": "heartbeat_runs",
|
||||
"columnsFrom": [
|
||||
"run_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"id"
|
||||
],
|
||||
"onDelete": "cascade",
|
||||
"onUpdate": "no action"
|
||||
}
|
||||
},
|
||||
"compositePrimaryKeys": {
|
||||
"heartbeat_run_event_cursors_company_id_run_id_source_instance_id_pk": {
|
||||
"name": "heartbeat_run_event_cursors_company_id_run_id_source_instance_id_pk",
|
||||
"columns": [
|
||||
"company_id",
|
||||
"run_id",
|
||||
"source_instance_id"
|
||||
]
|
||||
}
|
||||
},
|
||||
"uniqueConstraints": {},
|
||||
"policies": {},
|
||||
"checkConstraints": {
|
||||
"heartbeat_run_event_cursors_nonnegative": {
|
||||
"name": "heartbeat_run_event_cursors_nonnegative",
|
||||
"value": "\"heartbeat_run_event_cursors\".\"contiguous_source_seq\" >= 0 and \"heartbeat_run_event_cursors\".\"pending_range_count\" >= 0"
|
||||
}
|
||||
},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.heartbeat_run_event_source_ranges": {
|
||||
"name": "heartbeat_run_event_source_ranges",
|
||||
"schema": "",
|
||||
"columns": {
|
||||
"company_id": {
|
||||
"name": "company_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"run_id": {
|
||||
"name": "run_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"source_instance_id": {
|
||||
"name": "source_instance_id",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"start_seq": {
|
||||
"name": "start_seq",
|
||||
"type": "bigint",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"end_seq": {
|
||||
"name": "end_seq",
|
||||
"type": "bigint",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
}
|
||||
},
|
||||
"indexes": {},
|
||||
"foreignKeys": {
|
||||
"heartbeat_run_event_source_ranges_cursor_fk": {
|
||||
"name": "heartbeat_run_event_source_ranges_cursor_fk",
|
||||
"tableFrom": "heartbeat_run_event_source_ranges",
|
||||
"tableTo": "heartbeat_run_event_cursors",
|
||||
"columnsFrom": [
|
||||
"company_id",
|
||||
"run_id",
|
||||
"source_instance_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"company_id",
|
||||
"run_id",
|
||||
"source_instance_id"
|
||||
],
|
||||
"onDelete": "cascade",
|
||||
"onUpdate": "no action"
|
||||
}
|
||||
},
|
||||
"compositePrimaryKeys": {
|
||||
"heartbeat_run_event_source_ranges_company_id_run_id_source_instance_id_start_seq_pk": {
|
||||
"name": "heartbeat_run_event_source_ranges_company_id_run_id_source_instance_id_start_seq_pk",
|
||||
"columns": [
|
||||
"company_id",
|
||||
"run_id",
|
||||
"source_instance_id",
|
||||
"start_seq"
|
||||
]
|
||||
}
|
||||
},
|
||||
"uniqueConstraints": {},
|
||||
"policies": {},
|
||||
"checkConstraints": {
|
||||
"heartbeat_run_event_source_ranges_positive": {
|
||||
"name": "heartbeat_run_event_source_ranges_positive",
|
||||
"value": "\"heartbeat_run_event_source_ranges\".\"start_seq\" >= 1 and \"heartbeat_run_event_source_ranges\".\"end_seq\" >= \"heartbeat_run_event_source_ranges\".\"start_seq\""
|
||||
}
|
||||
},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.heartbeat_run_events": {
|
||||
"name": "heartbeat_run_events",
|
||||
"schema": "",
|
||||
@@ -33873,6 +34338,20 @@
|
||||
"notNull": true,
|
||||
"default": 1
|
||||
},
|
||||
"event_cursor_projection_seq": {
|
||||
"name": "event_cursor_projection_seq",
|
||||
"type": "bigint",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": 0
|
||||
},
|
||||
"event_cursor_rebuild_pending": {
|
||||
"name": "event_cursor_rebuild_pending",
|
||||
"type": "boolean",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": false
|
||||
},
|
||||
"native_phase": {
|
||||
"name": "native_phase",
|
||||
"type": "text",
|
||||
@@ -34303,6 +34782,28 @@
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"heartbeat_runs_event_cursor_rebuild_idx": {
|
||||
"name": "heartbeat_runs_event_cursor_rebuild_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"heartbeat_runs\".\"event_cursor_rebuild_pending\" = true",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"heartbeat_runs_company_agent_started_idx": {
|
||||
"name": "heartbeat_runs_company_agent_started_idx",
|
||||
"columns": [
|
||||
@@ -50177,6 +50678,12 @@
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"created_by_run_id": {
|
||||
"name": "created_by_run_id",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"dispatch_fingerprint": {
|
||||
"name": "dispatch_fingerprint",
|
||||
"type": "text",
|
||||
@@ -62139,6 +62646,12 @@
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"cleanup_verified_at": {
|
||||
"name": "cleanup_verified_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"created_at": {
|
||||
"name": "created_at",
|
||||
"type": "timestamp with time zone",
|
||||
+7
@@ -2969,6 +2969,13 @@
|
||||
"when": 1793724587619,
|
||||
"tag": "0425_awesome_living_lightning",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 426,
|
||||
"version": "7",
|
||||
"when": 1793732465918,
|
||||
"tag": "0426_nappy_deadpool",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { bigint, check, foreignKey, integer, pgTable, primaryKey, text, uuid } from "drizzle-orm/pg-core";
|
||||
import { companies } from "./companies.js";
|
||||
import { heartbeatRuns } from "./heartbeat_runs.js";
|
||||
|
||||
/** Derived from retained run events; never an independent replay authority. */
|
||||
export const heartbeatRunEventCursors = pgTable("heartbeat_run_event_cursors", {
|
||||
companyId: uuid("company_id").notNull().references(() => companies.id),
|
||||
runId: uuid("run_id").notNull().references(() => heartbeatRuns.id, { onDelete: "cascade" }),
|
||||
sourceInstanceId: text("source_instance_id").notNull(),
|
||||
contiguousSourceSeq: bigint("contiguous_source_seq", { mode: "number" }).notNull().default(0),
|
||||
pendingRangeCount: integer("pending_range_count").notNull().default(0),
|
||||
}, table => ({
|
||||
pk: primaryKey({ columns: [table.companyId, table.runId, table.sourceInstanceId] }),
|
||||
nonnegative: check("heartbeat_run_event_cursors_nonnegative", sql`${table.contiguousSourceSeq} >= 0 and ${table.pendingRangeCount} >= 0`),
|
||||
}));
|
||||
|
||||
/** Nonadjacent received ranges strictly above the contiguous source prefix. */
|
||||
export const heartbeatRunEventSourceRanges = pgTable("heartbeat_run_event_source_ranges", {
|
||||
companyId: uuid("company_id").notNull(),
|
||||
runId: uuid("run_id").notNull(),
|
||||
sourceInstanceId: text("source_instance_id").notNull(),
|
||||
startSeq: bigint("start_seq", { mode: "number" }).notNull(),
|
||||
endSeq: bigint("end_seq", { mode: "number" }).notNull(),
|
||||
}, table => ({
|
||||
pk: primaryKey({ columns: [table.companyId, table.runId, table.sourceInstanceId, table.startSeq] }),
|
||||
cursor: foreignKey({ name: "heartbeat_run_event_source_ranges_cursor_fk",
|
||||
columns: [table.companyId, table.runId, table.sourceInstanceId],
|
||||
foreignColumns: [heartbeatRunEventCursors.companyId, heartbeatRunEventCursors.runId, heartbeatRunEventCursors.sourceInstanceId],
|
||||
}).onDelete("cascade"),
|
||||
positive: check("heartbeat_run_event_source_ranges_positive", sql`${table.startSeq} >= 1 and ${table.endSeq} >= ${table.startSeq}`),
|
||||
}));
|
||||
@@ -76,6 +76,9 @@ export const heartbeatRuns = pgTable(
|
||||
completionContractId: uuid("completion_contract_id"),
|
||||
completionContractSha256: text("completion_contract_sha256"),
|
||||
nextEventSeq: bigint("next_event_seq", { mode: "number" }).notNull().default(1),
|
||||
// Disposable source-cursor projection, valid only through this global seq.
|
||||
eventCursorProjectionSeq: bigint("event_cursor_projection_seq", { mode: "number" }).notNull().default(0),
|
||||
eventCursorRebuildPending: boolean("event_cursor_rebuild_pending").notNull().default(false),
|
||||
nativePhase: text("native_phase"),
|
||||
nativePhaseUpdatedAt: timestamp("native_phase_updated_at", { withTimezone: true }),
|
||||
sessionIdBefore: text("session_id_before"),
|
||||
@@ -156,6 +159,8 @@ export const heartbeatRuns = pgTable(
|
||||
table.nativeIssueId,
|
||||
table.id,
|
||||
),
|
||||
eventCursorRebuildIdx: index("heartbeat_runs_event_cursor_rebuild_idx")
|
||||
.on(table.companyId, table.id).where(sql`${table.eventCursorRebuildPending} = true`),
|
||||
companyNativeIssueRunContractUq: unique(
|
||||
"heartbeat_runs_company_native_issue_contract_id_uq",
|
||||
).on(
|
||||
|
||||
@@ -274,3 +274,4 @@ export { workflows } from "./workflows.js";
|
||||
export { userWorkspacePreferences } from "./user_workspace_preferences.js";
|
||||
|
||||
export { issueThreadInteractionIdempotencyKeys } from "./issue_thread_interaction_idempotency_keys.js";
|
||||
export { heartbeatRunEventCursors, heartbeatRunEventSourceRanges } from "./heartbeat_run_event_cursors.js";
|
||||
@@ -0,0 +1,365 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import { createRequire } from "node:module";
|
||||
import { availableParallelism, loadavg } from "node:os";
|
||||
import path from "node:path";
|
||||
import { monitorEventLoopDelay, performance } from "node:perf_hooks";
|
||||
import { fileURLToPath, pathToFileURL } from "node:url";
|
||||
import {
|
||||
agents, applyPendingMigrations, companies, createDb, heartbeatRuns,
|
||||
startEmbeddedPostgresTestDatabase, type Db,
|
||||
} from "../packages/db/src/index.js";
|
||||
import { installDatabaseWorkSignals } from "../packages/db/src/work-signals.js";
|
||||
import type { AppendHeartbeatRunEventInput, AppendHeartbeatRunEventResult } from "../server/src/services/heartbeat-run-events.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../server/src/__tests__/helpers/external-test-database.js";
|
||||
|
||||
// This measures the real database owner, not browser delivery or a provider.
|
||||
// The assembled application/browser acceptance is a separate required layer.
|
||||
const root = fileURLToPath(new URL("../", import.meta.url));
|
||||
const serverRequire = createRequire(new URL("../server/package.json", import.meta.url));
|
||||
const { drizzle } = await import(serverRequire.resolve("drizzle-orm/postgres-js"));
|
||||
const sourcePath = "server/src/services/heartbeat-run-events.ts";
|
||||
const arg = (name: string) => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3);
|
||||
const git = (args: string[]) => execFileSync("git", args, { cwd: root, encoding: "utf8" });
|
||||
if (arg("validate-report")) {
|
||||
const recorded = JSON.parse(await fs.readFile(arg("validate-report")!, "utf8"));
|
||||
assert.equal(recorded.incomplete, false, "Incomplete evidence cannot qualify a revision");
|
||||
assert.ok(Object.keys(recorded.candidateSourceHashes ?? {}).length > 0, "Report must include supporting source hashes");
|
||||
for (const [file, expected] of Object.entries(recorded.candidateSourceHashes)) {
|
||||
assert.equal(createHash("sha256").update(await fs.readFile(path.join(root, file))).digest("hex"), expected, `Measured source changed: ${file}`);
|
||||
}
|
||||
console.log(JSON.stringify({ valid: true, phase: recorded.phase, baselineRevision: recorded.baselineRevision,
|
||||
candidateRevision: recorded.candidateRevision, scope: recorded.scope }));
|
||||
process.exit(0);
|
||||
}
|
||||
const quick = process.argv.includes("--quick");
|
||||
const phase = arg("phase") ?? "compare";
|
||||
assert.ok(["aa", "compare", "batch"].includes(phase), "--phase must be aa, compare, or batch");
|
||||
const baselineRef = arg("baseline-ref");
|
||||
assert.ok(baselineRef, "Pass a pinned --baseline-ref=<commit>");
|
||||
const baselineRevision = git(["rev-parse", "--verify", `${baselineRef}^{commit}`]).trim();
|
||||
const baselineSource = git(["show", `${baselineRevision}:${sourcePath}`]);
|
||||
const candidateRevision = git(["rev-parse", "HEAD"]).trim();
|
||||
const histories = (arg("histories") ?? (quick ? "1000" : "1000,10000,100000")).split(",").map(Number);
|
||||
const pairs = Number(arg("pairs") ?? (quick ? 1 : 5));
|
||||
const events = Number(arg("events") ?? (quick ? 8 : 32));
|
||||
assert.ok(histories.every(value => Number.isSafeInteger(value) && value >= 1));
|
||||
assert.ok(Number.isSafeInteger(pairs) && pairs >= (quick ? 1 : 5), "Acceptance requires at least five paired samples");
|
||||
assert.ok(Number.isSafeInteger(events) && events >= 2 && events <= 64 && events % 2 === 0);
|
||||
const scenarios = quick ? ["ordered"] : ["ordered", "replay", "late-gap", "warm-suffix"];
|
||||
const workload = { histories, pairs, events, scenarios, payloadTextBytes: 128, warmSourceOffset: 1000,
|
||||
seed: "native-event-ingestion-v1", retainedHistory: "vacuum-analyzed", quick };
|
||||
const workloadSha256 = createHash("sha256").update(JSON.stringify(workload)).digest("hex");
|
||||
const output = arg("output");
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EVENT_INGESTION_BENCHMARK_DATABASE_URL");
|
||||
if (externalUrl) assert.ok(["127.0.0.1", "localhost"].includes(new URL(externalUrl).hostname), "Use a disposable local database");
|
||||
|
||||
type Append = (db: Db, input: AppendHeartbeatRunEventInput) => Promise<AppendHeartbeatRunEventResult>;
|
||||
type AppendBatch = (db: Db, inputs: AppendHeartbeatRunEventInput[]) => Promise<AppendHeartbeatRunEventResult[]>;
|
||||
type EventModule = { appendHeartbeatRunEvent: Append; appendHeartbeatRunEvents?: AppendBatch };
|
||||
type Fixture = { companyId: string; agentId: string; runId: string; source: string; history: number; sourceOffset: number; next: number };
|
||||
type Query = { sql: string; params: unknown[]; elapsedMs?: number; returnedRows?: number };
|
||||
type Sample = {
|
||||
elapsedMs: number; eventsPerSecond: number; newEvents: number; duplicateEvents: number;
|
||||
appCpuMicros: number; databaseCpuMillis: number | null; heapDeltaBytes: number; rssBytes: number;
|
||||
eventLoopUtilization: number; eventLoopDelayMaxMs: number | null;
|
||||
};
|
||||
type Arm = "baseline" | "candidate";
|
||||
const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!;
|
||||
const summary = (samples: Sample[]) => ({
|
||||
medianMs: median(samples.map(sample => sample.elapsedMs)),
|
||||
minMs: Math.min(...samples.map(sample => sample.elapsedMs)), maxMs: Math.max(...samples.map(sample => sample.elapsedMs)),
|
||||
medianEventsPerSecond: median(samples.map(sample => sample.eventsPerSecond)),
|
||||
medianAppCpuMicrosPerEvent: median(samples.map(sample => sample.appCpuMicros / (sample.newEvents + sample.duplicateEvents))),
|
||||
samples,
|
||||
});
|
||||
const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.native-event-baseline-"));
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: Db | undefined;
|
||||
const companiesToRemove: string[] = [];
|
||||
const setupStarted = performance.now();
|
||||
const loadBefore = loadavg();
|
||||
const report: Record<string, unknown> = {
|
||||
scope: "Real append service / real PostgreSQL; closed-loop service cost, not assembled application, browser, or hosted acceptance",
|
||||
phase, baselineRevision, candidateRevision, workload, workloadSha256,
|
||||
baselineSourceSha256: createHash("sha256").update(baselineSource).digest("hex"),
|
||||
measurement: { warmups: 2, pairedOrder: "alternating", cache: "warm; seed and projection preparation excluded", databaseCpuResolutionMillis: 10 },
|
||||
environment: { node: process.version, platform: process.platform, architecture: process.arch, cpus: availableParallelism(), loadBefore },
|
||||
results: [], incomplete: true,
|
||||
};
|
||||
|
||||
async function save() {
|
||||
if (output) await fs.writeFile(output, JSON.stringify(report, null, 2));
|
||||
}
|
||||
|
||||
function backendCpuMillis(pid: number): number | null {
|
||||
if (process.platform === "win32") return null;
|
||||
const value = execFileSync("ps", ["-o", "time=", "-p", String(pid)], { encoding: "utf8" }).trim();
|
||||
const parts = value.split(":").map(Number);
|
||||
return parts.length >= 2 && parts.every(Number.isFinite)
|
||||
? parts.reduce((total, part) => total * 60 + part, 0) * 1000 : null;
|
||||
}
|
||||
|
||||
function input(fixture: Fixture, sequence: number): AppendHeartbeatRunEventInput {
|
||||
const envelope = {
|
||||
schema: "paperclip.prp.event.v1", schemaVersion: 1, runId: fixture.runId,
|
||||
normalizedSessionId: fixture.runId, sourceKind: "runner", sourceInstanceId: fixture.source,
|
||||
sourceEventId: `${fixture.source}:${sequence}`, sourceSeq: sequence,
|
||||
eventType: "item.delta", priority: 1, emittedAt: "2026-11-03T00:00:00.000Z",
|
||||
turnId: "benchmark-turn", itemId: "benchmark-item",
|
||||
payload: { kind: "agentMessage", channel: "final", text: "x".repeat(workload.payloadTextBytes) },
|
||||
};
|
||||
return {
|
||||
companyId: fixture.companyId, agentId: fixture.agentId, runId: fixture.runId,
|
||||
eventType: envelope.eventType, stream: "system", level: "info", payload: { prpEvent: envelope },
|
||||
nativeSource: {
|
||||
sourceInstanceId: fixture.source, sourceEventId: envelope.sourceEventId, sourceSeq: sequence,
|
||||
protocolSchemaVersion: 1, canonicalPayload: envelope,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// Query diagnostics run separately from unprofiled timing. Wrapping the actual
|
||||
// prepared execution captures completed query duration, including transactions.
|
||||
function diagnosticDb(target: Db, queries: Query[], transactions: Array<{ elapsedMs: number; lockAcquisitionQueryMs?: number; lockHeldThroughCommitMs?: number }>): Db {
|
||||
const recorded = installDatabaseWorkSignals(drizzle(target.$client, { schema: target._.fullSchema }), target) as Db;
|
||||
type Session = { prepareQuery: (...args: any[]) => { execute: (...args: any[]) => Promise<unknown> } };
|
||||
const decorate = (connection: Db, transaction?: { acquisitionMs?: number; acquiredAt?: number }) => {
|
||||
const session = (connection as unknown as { session: Session }).session;
|
||||
const prepare = session.prepareQuery;
|
||||
session.prepareQuery = function (...args: any[]) {
|
||||
const prepared = prepare.apply(this, args), execute = prepared.execute;
|
||||
const query = args[0] as Query;
|
||||
prepared.execute = async function (...values: any[]) {
|
||||
const started = performance.now();
|
||||
try {
|
||||
const result = await execute.apply(this, values);
|
||||
const elapsedMs = performance.now() - started;
|
||||
queries.push({ sql: query.sql, params: query.params, elapsedMs, returnedRows: Array.isArray(result) ? result.length : undefined });
|
||||
if (transaction && /\bfor update\b/i.test(query.sql)) {
|
||||
transaction.acquisitionMs = elapsedMs;
|
||||
transaction.acquiredAt = performance.now();
|
||||
}
|
||||
return result;
|
||||
} catch (error) {
|
||||
queries.push({ sql: query.sql, params: query.params, elapsedMs: performance.now() - started });
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
return prepared;
|
||||
};
|
||||
const original = connection.transaction.bind(connection);
|
||||
connection.transaction = ((callback: (tx: Db) => Promise<unknown>, ...options: any[]) => {
|
||||
const started = performance.now(), state: { acquisitionMs?: number; acquiredAt?: number } = {};
|
||||
return original(async tx => callback(decorate(tx as unknown as Db, state)), ...options).finally(() => {
|
||||
transactions.push({ elapsedMs: performance.now() - started, lockAcquisitionQueryMs: state.acquisitionMs,
|
||||
lockHeldThroughCommitMs: state.acquiredAt === undefined ? undefined : performance.now() - state.acquiredAt });
|
||||
});
|
||||
}) as typeof connection.transaction;
|
||||
return connection;
|
||||
};
|
||||
return decorate(recorded);
|
||||
}
|
||||
|
||||
try {
|
||||
// Snapshot changed supporting service modules too: a B1→B2 comparison must
|
||||
// not accidentally import B2 behavior behind its nominally B1 entry point.
|
||||
// The database schema/client intentionally remain the compatible migrated
|
||||
// superset for both arms; the benchmark is about the event owner itself.
|
||||
const changed = git(["diff", "--name-only", baselineRevision, "--", "server/src/services"])
|
||||
.trim().split("\n").filter(file => file.endsWith(".ts") && !file.endsWith(".test.ts"));
|
||||
const snapshots = new Map<string, string>([[sourcePath, baselineSource]]);
|
||||
for (const file of changed) {
|
||||
if (snapshots.has(file)) continue;
|
||||
try { snapshots.set(file, git(["show", `${baselineRevision}:${file}`])); }
|
||||
catch { /* A newly added service cannot have been imported by the baseline. */ }
|
||||
}
|
||||
for (const [file, source] of snapshots) {
|
||||
const destination = path.join(baselineDir, file);
|
||||
await fs.mkdir(path.dirname(destination), { recursive: true });
|
||||
const sourceUrl = pathToFileURL(path.join(root, file));
|
||||
await fs.writeFile(destination, source.replace(/from "(\.{1,2}\/[^\"]+)"/g, (_match, specifier: string) => {
|
||||
const actual = fileURLToPath(new URL(specifier, sourceUrl));
|
||||
const relative = path.relative(root, actual).replace(/\.js$/, ".ts");
|
||||
return `from ${JSON.stringify(snapshots.has(relative) ? path.join(baselineDir, relative) : actual)}`;
|
||||
}));
|
||||
}
|
||||
report.baselineServiceSnapshots = Object.fromEntries([...snapshots].map(([file, source]) => [file, createHash("sha256").update(source).digest("hex")]));
|
||||
const baselineFile = path.join(baselineDir, sourcePath);
|
||||
const baseline = await import(pathToFileURL(baselineFile).href) as EventModule;
|
||||
const candidate = phase === "aa" ? baseline : await import("../server/src/services/heartbeat-run-events.js") as EventModule;
|
||||
if (phase === "batch") assert.equal(typeof candidate.appendHeartbeatRunEvents, "function", "Batch phase requires appendHeartbeatRunEvents");
|
||||
report.candidateSourceSha256 = createHash("sha256").update(await fs.readFile(path.join(root, sourcePath))).digest("hex");
|
||||
const proofFiles = [...new Set([sourcePath, "scripts/benchmark-native-event-ingestion.ts",
|
||||
...git(["diff", "--name-only", baselineRevision, "--", "server/src/services", "packages/db/src/schema"]).trim().split("\n"),
|
||||
...git(["ls-files", "--others", "--exclude-standard", "--", "server/src/services", "packages/db/src/schema"]).trim().split("\n"),
|
||||
])].filter(file => file.endsWith(".ts") && !file.endsWith(".test.ts"));
|
||||
report.candidateSourceHashes = Object.fromEntries(await Promise.all(proofFiles.map(async file => [file,
|
||||
createHash("sha256").update(await fs.readFile(path.join(root, file))).digest("hex")])));
|
||||
report.candidateDirtyDiff = git(["diff", "HEAD", "--", sourcePath, "server/src/services/heartbeat", "packages/db/src/schema"]);
|
||||
if (!externalUrl) database = await startEmbeddedPostgresTestDatabase("native-event-ingestion-benchmark-");
|
||||
if (externalUrl) await applyPendingMigrations(externalUrl);
|
||||
db = createDb(externalUrl ?? database!.connectionString, { maxConnections: 1 });
|
||||
const [settings] = await db.$client.unsafe("select version() postgres_version, pg_backend_pid() backend_pid, current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit");
|
||||
report.database = settings;
|
||||
|
||||
async function fixture(history: number, scenario = "ordered"): Promise<Fixture> {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), source = randomUUID();
|
||||
companiesToRemove.push(companyId);
|
||||
await db!.insert(companies).values({ id: companyId, name: "Synthetic native event benchmark", issuePrefix: `E${companyId.replaceAll("-", "")}` });
|
||||
await db!.insert(agents).values({ id: agentId, companyId, name: "Synthetic event source", adapterType: "paperclip_runner" });
|
||||
await db!.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", nextEventSeq: history + 1 });
|
||||
const sourceOffset = scenario === "late-gap" ? 1 : scenario === "warm-suffix" ? workload.warmSourceOffset : 0;
|
||||
await db!.$client.unsafe(`insert into heartbeat_run_events
|
||||
(company_id, run_id, agent_id, seq, event_type, stream, level, payload, source_instance_id, source_event_id, source_seq, source_payload_sha256, protocol_schema_version)
|
||||
select $1::uuid, $2::uuid, $3::uuid, n, 'item.delta', 'system', 'info', jsonb_build_object('historicalSyntheticText', repeat('x', $5::integer)),
|
||||
$4, $4 || ':' || (n + $6::integer)::text, n + $6::integer, repeat('0', 64), 1
|
||||
from generate_series(1, $7::integer) n`, [companyId, runId, agentId, source, workload.payloadTextBytes, sourceOffset, history]);
|
||||
await db!.$client.unsafe("VACUUM ANALYZE heartbeat_run_events");
|
||||
return { companyId, agentId, runId, source, history, sourceOffset, next: history + sourceOffset + 1 };
|
||||
}
|
||||
|
||||
async function prepare(f: Fixture, arm: Arm) {
|
||||
if (arm === "baseline" || phase === "aa") return;
|
||||
// Seed setup is an old-version append. Prime using the real fallback and
|
||||
// bounded rebuild before steady-state timing; cold behavior is traced below.
|
||||
const module = await import("../server/src/services/heartbeat/native-event-cursors.js") as Record<string, unknown>;
|
||||
const rebuild = module.rebuildNativeEventCursorPage as undefined | ((db: Db, binding: { companyId: string; runId: string }) => Promise<{ pending: boolean }>);
|
||||
assert.equal(typeof rebuild, "function", "Cursor owner must expose rebuildNativeEventCursorPage for seeded-history preparation");
|
||||
await db!.$client.unsafe("update heartbeat_runs set event_cursor_rebuild_pending=true where id=$1::uuid", [f.runId]);
|
||||
for (let page = 0; page <= Math.ceil((f.history + 1) / 512) + 2; page++) {
|
||||
const result = await rebuild!(db!, { companyId: f.companyId, runId: f.runId });
|
||||
if (!result.pending) return;
|
||||
}
|
||||
throw new Error("Cursor fixture failed to finish bounded rebuild");
|
||||
}
|
||||
|
||||
function emittedInputs(f: Fixture, scenario: string) {
|
||||
if (scenario === "late-gap") return [input(f, 1)];
|
||||
const newCount = scenario === "replay" ? events / 2 : events;
|
||||
const unique = Array.from({ length: newCount }, (_, n) => input(f, f.next + n));
|
||||
f.next += newCount;
|
||||
return scenario === "replay" ? [...unique, ...unique] : unique;
|
||||
}
|
||||
|
||||
async function invoke(target: Db, arm: Arm, values: AppendHeartbeatRunEventInput[]) {
|
||||
const owner = arm === "baseline" ? baseline : candidate;
|
||||
const receipts: AppendHeartbeatRunEventResult[] = [];
|
||||
if (phase === "batch" && arm === "candidate") receipts.push(...await owner.appendHeartbeatRunEvents!(target, values));
|
||||
else for (const value of values) receipts.push(await owner.appendHeartbeatRunEvent(target, value));
|
||||
assert.equal(receipts.length, values.length);
|
||||
const expectedCursor = Math.max(...values.map(value => value.nativeSource!.sourceSeq));
|
||||
assert.ok(receipts.every(receipt => receipt.highestContiguousSourceSeq <= expectedCursor || values.length === 1));
|
||||
return receipts;
|
||||
}
|
||||
|
||||
async function timed(f: Fixture, arm: Arm, scenario: string): Promise<Sample> {
|
||||
const values = emittedInputs(f, scenario);
|
||||
const pgBefore = backendCpuMillis(Number(settings!.backend_pid));
|
||||
const beforeMemory = process.memoryUsage(), cpuBefore = process.cpuUsage(), eluBefore = performance.eventLoopUtilization();
|
||||
const loop = monitorEventLoopDelay({ resolution: 10 });
|
||||
loop.enable();
|
||||
const started = performance.now();
|
||||
const receipts = await invoke(db!, arm, values);
|
||||
const elapsedMs = performance.now() - started;
|
||||
loop.disable();
|
||||
const cpu = process.cpuUsage(cpuBefore), memory = process.memoryUsage();
|
||||
const pgAfter = backendCpuMillis(Number(settings!.backend_pid));
|
||||
const newEvents = receipts.filter(receipt => receipt.disposition === "committed").length;
|
||||
assert.equal(newEvents, scenario === "replay" ? events / 2 : values.length);
|
||||
assert.equal(receipts.at(-1)!.highestContiguousSourceSeq,
|
||||
scenario === "late-gap" ? f.history + 1 : scenario === "warm-suffix" ? 0 : f.next - 1);
|
||||
return { elapsedMs, eventsPerSecond: values.length / elapsedMs * 1000, newEvents,
|
||||
duplicateEvents: values.length - newEvents, appCpuMicros: cpu.user + cpu.system,
|
||||
databaseCpuMillis: pgBefore === null || pgAfter === null ? null : pgAfter - pgBefore,
|
||||
heapDeltaBytes: memory.heapUsed - beforeMemory.heapUsed, rssBytes: memory.rss,
|
||||
eventLoopUtilization: performance.eventLoopUtilization(eluBefore).utilization,
|
||||
eventLoopDelayMaxMs: loop.count ? loop.max / 1e6 : null };
|
||||
}
|
||||
|
||||
// Qualify before entering the full benchmark. All comparisons preserve the
|
||||
// original append receipts; a failed fixture never becomes a timing sample.
|
||||
for (const history of histories) for (const scenario of scenarios) {
|
||||
const samples: Record<Arm, Sample[]> = { baseline: [], candidate: [] };
|
||||
let a: Fixture | undefined, b: Fixture | undefined;
|
||||
if (scenario !== "late-gap") {
|
||||
a = await fixture(history, scenario); b = await fixture(history, scenario); await prepare(b, "candidate");
|
||||
for (let warmup = 0; warmup < 2; warmup++) {
|
||||
await timed(a, "baseline", scenario); await timed(b, "candidate", scenario);
|
||||
}
|
||||
}
|
||||
for (let pair = 0; pair < pairs; pair++) {
|
||||
if (scenario === "late-gap") {
|
||||
a = await fixture(history, scenario); b = await fixture(history, scenario); await prepare(b, "candidate");
|
||||
}
|
||||
for (const arm of pair % 2 ? ["candidate", "baseline"] as const : ["baseline", "candidate"] as const) {
|
||||
samples[arm].push(await timed(arm === "baseline" ? a! : b!, arm, scenario));
|
||||
}
|
||||
}
|
||||
const diagnostics: Record<string, unknown> = {};
|
||||
for (const arm of ["baseline", "candidate"] as const) {
|
||||
const f = await fixture(history, scenario); await prepare(f, arm);
|
||||
const queries: Query[] = [], transactions: Array<{ elapsedMs: number }> = [];
|
||||
const recorded = diagnosticDb(db, queries, transactions);
|
||||
const values = emittedInputs(f, scenario), receipts = await invoke(recorded, arm, values);
|
||||
const plans: unknown[] = [];
|
||||
for (const query of queries) {
|
||||
if (plans.length >= 16 || plans.some((plan: any) => plan.sql === query.sql)) continue;
|
||||
const read = /^select\b/i.test(query.sql);
|
||||
// Never reexecute a mutation merely to profile it. Read plans report
|
||||
// actual rows/buffers; mutation plans describe shape without effects.
|
||||
const [plan] = await db.$client.unsafe(`EXPLAIN (${read ? "ANALYZE, BUFFERS, " : ""}FORMAT JSON) ${query.sql}`, query.params as never[]);
|
||||
plans.push({ sql: query.sql, analyzed: read, plan: plan!["QUERY PLAN"] });
|
||||
}
|
||||
diagnostics[arm] = { sqlCalls: queries.length, transactions: transactions.length, sqlCallsPerEvent: queries.length / values.length,
|
||||
transactionsPerEvent: transactions.length / values.length, transactionTimings: transactions,
|
||||
lockTimingBoundary: "Acquisition query duration includes network/row-read cost; held interval ends after commit. It is not a measurement of pure lock wait.",
|
||||
persistedEvents: receipts.filter(receipt => receipt.disposition === "committed").length, queries, plans };
|
||||
}
|
||||
const baselineStats = summary(samples.baseline), candidateStats = summary(samples.candidate);
|
||||
const relativePairedDifference = samples.baseline.map((sample, n) => (samples.candidate[n]!.elapsedMs - sample.elapsedMs) / sample.elapsedMs);
|
||||
(report.results as unknown[]).push({ history, scenario, baseline: baselineStats, candidate: candidateStats, diagnostics,
|
||||
pairedCandidateWins: relativePairedDifference.filter(value => value < 0).length,
|
||||
medianLatencyChangePercent: 100 * (candidateStats.medianMs / baselineStats.medianMs - 1),
|
||||
aaNoiseEnvelopeRelative: phase === "aa" ? Math.max(...relativePairedDifference.map(Math.abs)) : undefined,
|
||||
});
|
||||
await save();
|
||||
}
|
||||
report.setupAndExecutionMillis = performance.now() - setupStarted;
|
||||
report.loadAfter = loadavg();
|
||||
report.incomplete = false;
|
||||
report.acceptance = phase === "aa" ? "Baseline repeatability only; this does not establish an improvement" : "Compare against the frozen same-workload A/A report; throughput improvement needs at least four of five paired wins beyond baseline noise. Browser, control and full-app gates remain separate.";
|
||||
if (arg("noise-report")) {
|
||||
const noise = JSON.parse(await fs.readFile(arg("noise-report")!, "utf8"));
|
||||
assert.equal(noise.phase, "aa"); assert.equal(noise.incomplete, false);
|
||||
assert.equal(noise.baselineRevision, baselineRevision); assert.equal(noise.workloadSha256, workloadSha256);
|
||||
report.noiseReport = arg("noise-report");
|
||||
report.comparison = (report.results as any[]).map(result => {
|
||||
const control = noise.results.find((row: any) => row.history === result.history && row.scenario === result.scenario);
|
||||
assert.ok(control);
|
||||
const improvement = 1 - result.candidate.medianMs / result.baseline.medianMs;
|
||||
return { history: result.history, scenario: result.scenario, pairedCandidateWins: result.pairedCandidateWins,
|
||||
improvementRelative: improvement, noiseEnvelopeRelative: control.aaNoiseEnvelopeRelative,
|
||||
improvedBeyondNoise: result.pairedCandidateWins >= Math.ceil(pairs * 0.8) && improvement > control.aaNoiseEnvelopeRelative };
|
||||
});
|
||||
}
|
||||
await save();
|
||||
console.log(JSON.stringify(report, null, 2));
|
||||
} finally {
|
||||
try {
|
||||
for (const companyId of companiesToRemove) {
|
||||
await db?.$client.unsafe("delete from heartbeat_run_events where company_id=$1::uuid", [companyId]);
|
||||
await db?.$client.unsafe("delete from heartbeat_runs where company_id=$1::uuid", [companyId]);
|
||||
await db?.$client.unsafe("delete from agents where company_id=$1::uuid", [companyId]);
|
||||
await db?.$client.unsafe("delete from companies where id=$1::uuid", [companyId]);
|
||||
}
|
||||
} finally {
|
||||
try { await db?.$client.end({ timeout: 1 }); }
|
||||
finally {
|
||||
try { await database?.cleanup(); }
|
||||
finally { await fs.rm(baselineDir, { recursive: true, force: true }); }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -148,7 +148,9 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
|
||||
query.includes('from "heartbeat_runs"') && query.endsWith("for update"));
|
||||
expect(lockReads).toHaveLength(2);
|
||||
for (const query of lockReads) {
|
||||
expect(query).toMatch(/^select "company_id", "agent_id" from "heartbeat_runs" /);
|
||||
expect(query).toMatch(kind === "native"
|
||||
? /^select "company_id", "agent_id", "next_event_seq", "event_cursor_projection_seq", "event_cursor_rebuild_pending" from "heartbeat_runs" /
|
||||
: /^select "company_id", "agent_id" from "heartbeat_runs" /);
|
||||
expect(query).not.toMatch(/context_snapshot|runner_profile_json|result_json|stdout_excerpt/);
|
||||
}
|
||||
});
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { drizzle } from "drizzle-orm/postgres-js";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, companies, createDb, heartbeatRunEventCursors, heartbeatRunEventSourceRanges,
|
||||
heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
import { appendHeartbeatRunEvent, type AppendHeartbeatRunEventInput } from "../services/heartbeat-run-events.js";
|
||||
import { rebuildNativeEventCursorPage } from "../services/heartbeat/native-event-cursors.js";
|
||||
import { nativeEventCursorWork } from "../services/heartbeat/native-event-cursor-work.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { nativeSha256 } from "../services/native-runtime/canonical.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EVENT_CURSORS_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("transactional native event cursors", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let otherDb: ReturnType<typeof createDb>;
|
||||
let companyId: string, agentId: string, runId: string;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
const sourceInstanceId = "source";
|
||||
const errors: unknown[] = [];
|
||||
beforeAll(async () => {
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-event-cursors-");
|
||||
const url = external ?? temporary!.connectionString;
|
||||
db = createDb(url, { idleTimeoutSeconds: 0 }); otherDb = createDb(url, { idleTimeoutSeconds: 0 });
|
||||
}, 60_000);
|
||||
beforeEach(async () => {
|
||||
companyId = randomUUID(); agentId = randomUUID(); runId = randomUUID(); errors.length = 0;
|
||||
await db.insert(companies).values({ id: companyId, name: "Cursor test", issuePrefix: "CUR" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Cursor agent" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running" });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks();
|
||||
await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.companyId, companyId));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
|
||||
await db.delete(agents).where(eq(agents.companyId, companyId));
|
||||
await db.delete(companies).where(eq(companies.id, companyId));
|
||||
});
|
||||
afterAll(async () => { await otherDb?.$client.end({ timeout: 1 }); await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
function input(sourceSeq: number, source = sourceInstanceId): AppendHeartbeatRunEventInput {
|
||||
return { companyId, agentId, runId, eventType: "item.delta",
|
||||
nativeSource: { sourceInstanceId: source, sourceEventId: `${source}:${sourceSeq}`, sourceSeq,
|
||||
protocolSchemaVersion: 1, canonicalPayload: { value: sourceSeq } } };
|
||||
}
|
||||
const run = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]!;
|
||||
const cursor = async () => (await db.select().from(heartbeatRunEventCursors)
|
||||
.where(and(eq(heartbeatRunEventCursors.runId, runId), eq(heartbeatRunEventCursors.sourceInstanceId, sourceInstanceId))))[0];
|
||||
const ranges = () => db.select().from(heartbeatRunEventSourceRanges)
|
||||
.where(eq(heartbeatRunEventSourceRanges.runId, runId)).orderBy(heartbeatRunEventSourceRanges.startSeq);
|
||||
async function rebuild() {
|
||||
let result;
|
||||
do { result = await rebuildNativeEventCursorPage(db, { companyId, runId }); } while (result.pending);
|
||||
}
|
||||
async function seedGapHistory(last: number) {
|
||||
await db.execute(sql`insert into heartbeat_run_events
|
||||
(company_id, run_id, agent_id, seq, event_type, source_instance_id, source_event_id, source_seq)
|
||||
select ${companyId}::uuid, ${runId}::uuid, ${agentId}::uuid, value - 1, 'item.delta',
|
||||
${sourceInstanceId}, 'seed:' || value, value from generate_series(2, ${last}::bigint) value`);
|
||||
await db.update(heartbeatRuns).set({ nextEventSeq: last, eventCursorRebuildPending: true })
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
}
|
||||
|
||||
it("merges gaps and keeps every event with constant-size source metadata", async () => {
|
||||
const results = [];
|
||||
for (const seq of [2, 5, 6, 1, 3, 4]) results.push((await appendHeartbeatRunEvent(db, input(seq))).highestContiguousSourceSeq);
|
||||
expect(results).toEqual([0, 0, 0, 2, 3, 6]);
|
||||
expect(await ranges()).toEqual([]);
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 6, pendingRangeCount: 0 });
|
||||
expect(await run()).toMatchObject({ nextEventSeq: 7, eventCursorProjectionSeq: 6, eventCursorRebuildPending: false });
|
||||
expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId))).toHaveLength(6);
|
||||
expect((await appendHeartbeatRunEvent(otherDb, input(4))).highestContiguousSourceSeq).toBe(6);
|
||||
});
|
||||
|
||||
it("rejects collisions on either identity even when canonical hashes match", async () => {
|
||||
await appendHeartbeatRunEvent(db, input(1));
|
||||
await appendHeartbeatRunEvent(db, input(2));
|
||||
const sameDigest = input(1);
|
||||
await expect(appendHeartbeatRunEvent(db, { ...sameDigest,
|
||||
nativeSource: { ...sameDigest.nativeSource!, sourceSeq: 2 } })).rejects.toThrow("native_event_replay_conflict");
|
||||
await expect(appendHeartbeatRunEvent(db, { ...sameDigest,
|
||||
nativeSource: { ...sameDigest.nativeSource!, sourceInstanceId: "other" } })).rejects.toThrow("native_event_replay_conflict");
|
||||
expect((await run()).nextEventSeq).toBe(3);
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 2 });
|
||||
});
|
||||
|
||||
it("extends a warm-run suffix above a missing prefix in one projection statement", async () => {
|
||||
const first = 3_000_000_000;
|
||||
await appendHeartbeatRunEvent(db, input(first));
|
||||
const statements: string[] = [];
|
||||
const observed = drizzle(db.$client, { schema: db._.fullSchema,
|
||||
logger: { logQuery: query => statements.push(query) } }) as typeof db;
|
||||
expect((await appendHeartbeatRunEvent(observed, input(first + 1))).highestContiguousSourceSeq).toBe(0);
|
||||
expect(statements.filter(query => query.includes("heartbeat_run_event_cursors"))).toHaveLength(1);
|
||||
expect(await ranges()).toEqual([expect.objectContaining({ startSeq: first, endSeq: first + 1 })]);
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 0, pendingRangeCount: 1 });
|
||||
});
|
||||
|
||||
it("closes a large historical gap without counting or loading its event suffix", async () => {
|
||||
await seedGapHistory(10_000); await rebuild();
|
||||
expect(await ranges()).toEqual([expect.objectContaining({ startSeq: 2, endSeq: 10_000 })]);
|
||||
const statements: string[] = [];
|
||||
const observed = drizzle(db.$client, { schema: db._.fullSchema,
|
||||
logger: { logQuery: query => statements.push(query) } }) as typeof db;
|
||||
expect((await appendHeartbeatRunEvent(observed, input(1))).highestContiguousSourceSeq).toBe(10_000);
|
||||
expect(statements.some(query => /count\(\*\)|max\("source_seq"\)/i.test(query))).toBe(false);
|
||||
// The only event-history SELECT is the bounded identity lookup.
|
||||
expect(statements.filter(query => query.startsWith("select ") && query.includes('from "heartbeat_run_events"')))
|
||||
.toEqual([expect.stringMatching(/limit \$\d+$/)]);
|
||||
expect(await ranges()).toEqual([]);
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 10_000, pendingRangeCount: 0 });
|
||||
}, 60_000);
|
||||
|
||||
it("uses the correct fallback during partial rebuild and resumes after an older writer", async () => {
|
||||
await seedGapHistory(1026);
|
||||
expect(await rebuildNativeEventCursorPage(db, { companyId, runId })).toMatchObject({ pending: true, processed: 512 });
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 0 });
|
||||
expect((await appendHeartbeatRunEvent(db, input(1))).highestContiguousSourceSeq).toBe(1026);
|
||||
expect((await run()).eventCursorProjectionSeq).toBe(512);
|
||||
await rebuild();
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 1026 });
|
||||
const old = input(1028);
|
||||
await db.transaction(async tx => {
|
||||
const [allocated] = await tx.update(heartbeatRuns).set({ nextEventSeq: sql`${heartbeatRuns.nextEventSeq} + 1` })
|
||||
.where(eq(heartbeatRuns.id, runId)).returning({ seq: heartbeatRuns.nextEventSeq });
|
||||
await tx.insert(heartbeatRunEvents).values({ companyId, agentId, runId, seq: allocated.seq - 1,
|
||||
eventType: old.eventType, sourceInstanceId, sourceEventId: old.nativeSource!.sourceEventId,
|
||||
sourceSeq: 1028, sourcePayloadSha256: nativeSha256(old.nativeSource!.canonicalPayload) });
|
||||
});
|
||||
expect((await appendHeartbeatRunEvent(db, input(1027))).highestContiguousSourceSeq).toBe(1028);
|
||||
expect((await run()).eventCursorRebuildPending).toBe(true);
|
||||
await rebuild();
|
||||
expect(await cursor()).toMatchObject({ contiguousSourceSeq: 1028, pendingRangeCount: 0 });
|
||||
expect((await run()).eventCursorRebuildPending).toBe(false);
|
||||
});
|
||||
|
||||
it("rolls source state and rebuild intent back with the outer transaction", async () => {
|
||||
await expect(db.transaction(async tx => {
|
||||
await appendHeartbeatRunEvent(tx as unknown as typeof db, input(1)); throw new Error("outer rollback");
|
||||
})).rejects.toThrow("outer rollback");
|
||||
expect(await cursor()).toBeUndefined(); expect((await run()).nextEventSeq).toBe(1);
|
||||
await seedGapHistory(4);
|
||||
await db.update(heartbeatRuns).set({ eventCursorRebuildPending: false }).where(eq(heartbeatRuns.id, runId));
|
||||
await expect(db.transaction(async tx => {
|
||||
await appendHeartbeatRunEvent(tx as unknown as typeof db, input(1)); throw new Error("outer rollback");
|
||||
})).rejects.toThrow("outer rollback");
|
||||
expect((await run()).eventCursorRebuildPending).toBe(false);
|
||||
expect((await run()).nextEventSeq).toBe(4);
|
||||
});
|
||||
|
||||
it("reconstructs explicit pending work, becomes quiet, and reactivates after a stale append", async () => {
|
||||
await seedGapHistory(1026);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true,
|
||||
onError: error => errors.push(error) });
|
||||
await nativeEventCursorWork(db).register(coordinator, () => true).ready;
|
||||
const settle = () => vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect((await run()).eventCursorRebuildPending).toBe(false);
|
||||
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
await settle();
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
select.mockRestore(); transaction.mockRestore();
|
||||
// An old binary adds a legacy event without maintaining the marker.
|
||||
await db.execute(sql`with allocated as (update heartbeat_runs set next_event_seq = next_event_seq + 1
|
||||
where id = ${runId} returning next_event_seq - 1 as seq)
|
||||
insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type)
|
||||
select ${companyId}::uuid, ${runId}::uuid, ${agentId}::uuid, seq, 'log' from allocated`);
|
||||
await appendHeartbeatRunEvent(db, input(1));
|
||||
await settle(); expect(await cursor()).toMatchObject({ contiguousSourceSeq: 1026 }); expect(errors).toEqual([]);
|
||||
await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(await db.select().from(heartbeatRunEventCursors).where(eq(heartbeatRunEventCursors.runId, runId))).toEqual([]);
|
||||
expect(await ranges()).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -249,6 +249,15 @@ if (!support.supported) console.warn(`Skipping idle sleep Postgres tests: ${supp
|
||||
expect(await read()).toEqual(none);
|
||||
});
|
||||
|
||||
it("retains an explicitly pending event cursor rebuild until it settles", async () => {
|
||||
const f = await seed();
|
||||
expect(await read()).toEqual(none);
|
||||
await db.update(heartbeatRuns).set({ eventCursorRebuildPending: true }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await read()).toEqual(present);
|
||||
await db.update(heartbeatRuns).set({ eventCursorRebuildPending: false }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await read()).toEqual(none);
|
||||
});
|
||||
|
||||
async function feedbackHistory(status: string, change: Partial<typeof feedbackExports.$inferInsert> = {}) {
|
||||
const f = await seed();
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Finished fixture", status: "done" }).returning();
|
||||
|
||||
@@ -209,6 +209,7 @@ import { toolAccessService } from "./services/tool-access.js";
|
||||
import { chatChannelService } from "./services/chat-channels.js";
|
||||
import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js";
|
||||
import { nativeChatProgressWork } from "./services/chat-native-progress-work.js";
|
||||
import { nativeEventCursorWork } from "./services/heartbeat/native-event-cursor-work.js";
|
||||
import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js";
|
||||
import { chatFailedRetryNoticeWork } from "./services/chat-failed-retry-notice-work.js";
|
||||
import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js";
|
||||
@@ -1259,6 +1260,7 @@ export async function createApp(
|
||||
chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
nativeEventCursorWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatFailedRetryNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatRunMilestoneWork(db, opts.authPublicBaseUrl).register(
|
||||
|
||||
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db/work-
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
nativeEventCursorRebuild: "native-event-cursor-rebuild",
|
||||
nativeRuntimeResponses: "native-runtime-responses",
|
||||
nativeCancellation: "native-cancellation",
|
||||
deploymentNotifications: "deployment-notifications",
|
||||
|
||||
@@ -6,10 +6,13 @@ import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./na
|
||||
import { isSafeNativeChatProgressEventType } from "./safe-native-chat-progress.js";
|
||||
import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
import { and, asc, eq, getTableColumns, or, sql } from "drizzle-orm";
|
||||
import { and, eq, getTableColumns, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
import { nativeSha256 } from "./native-runtime/canonical.js";
|
||||
import { nativeEventCursorProjectionCurrent, projectNativeEventSourceRange, readNativeEventSourceCursor,
|
||||
readRetainedNativeEventCursor, requestNativeEventCursorRebuild, type NativeEventCursorRun,
|
||||
} from "./heartbeat/native-event-cursors.js";
|
||||
|
||||
export const RETRY_EXHAUSTION_MESSAGE_PREFIX = "Bounded retry exhausted";
|
||||
|
||||
@@ -57,15 +60,22 @@ export class HeartbeatRunEventConflictError extends Error {
|
||||
* ingestion and legacy/direct-adapter writers use this allocator so the
|
||||
* database uniqueness invariant cannot turn a concurrent log/cancel/recovery
|
||||
* race into a failed run.
|
||||
* Set projectionApplied only for a legacy event or after incorporating its
|
||||
* native source metadata in this same transaction. Otherwise leave the marker
|
||||
* stale so a subsequent native reader safely falls back and requests rebuild.
|
||||
*/
|
||||
export async function allocateHeartbeatRunEventSeq(
|
||||
db: Db,
|
||||
runId: string,
|
||||
projectionApplied = false,
|
||||
): Promise<number> {
|
||||
const [updated] = await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
nextEventSeq: sql`${heartbeatRuns.nextEventSeq} + 1`,
|
||||
...(projectionApplied ? { eventCursorProjectionSeq: sql`case
|
||||
when ${heartbeatRuns.eventCursorProjectionSeq} = ${heartbeatRuns.nextEventSeq} - 1
|
||||
then ${heartbeatRuns.nextEventSeq} else ${heartbeatRuns.eventCursorProjectionSeq} end` } : {}),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
@@ -87,6 +97,9 @@ export async function appendHeartbeatRunEvent(
|
||||
const allocated = db.$with("allocated_run_event", {}).as(sql`
|
||||
update ${heartbeatRuns}
|
||||
set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1,
|
||||
event_cursor_projection_seq = case
|
||||
when ${heartbeatRuns.eventCursorProjectionSeq} = ${heartbeatRuns.nextEventSeq} - 1
|
||||
then ${heartbeatRuns.nextEventSeq} else ${heartbeatRuns.eventCursorProjectionSeq} end,
|
||||
updated_at = ${sql.param(new Date(), heartbeatRuns.updatedAt)}
|
||||
where ${heartbeatRuns.id} = ${input.runId}
|
||||
and ${heartbeatRuns.companyId} = ${input.companyId}
|
||||
@@ -110,6 +123,9 @@ export async function appendHeartbeatRunEvent(
|
||||
return db.transaction(async (tx) => {
|
||||
const run = await tx
|
||||
.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
...(input.nativeSource ? { nextEventSeq: heartbeatRuns.nextEventSeq,
|
||||
eventCursorProjectionSeq: heartbeatRuns.eventCursorProjectionSeq,
|
||||
eventCursorRebuildPending: heartbeatRuns.eventCursorRebuildPending } : {}),
|
||||
// Only progress needs the task binding. Other native/recovery events
|
||||
// must not read a potentially large context snapshot to append a log.
|
||||
...(projectsProgress || projectsResponse || projectsReassessment ? { runtimeMode: heartbeatRuns.runtimeMode } : {}),
|
||||
@@ -123,6 +139,10 @@ export async function appendHeartbeatRunEvent(
|
||||
throw new Error("heartbeat_run_event_binding_mismatch");
|
||||
}
|
||||
|
||||
const cursorRun = input.nativeSource ? { ...run, id: input.runId } as NativeEventCursorRun : null;
|
||||
const sourceScope = input.nativeSource ? { companyId: input.companyId, runId: input.runId,
|
||||
sourceInstanceId: input.nativeSource.sourceInstanceId } : null;
|
||||
const projectionCurrent = cursorRun !== null && nativeEventCursorProjectionCurrent(cursorRun);
|
||||
if (projectsRecovery) await notifyLiveRunRecovery(tx, input);
|
||||
|
||||
if (input.retryExhaustion && !input.nativeSource) {
|
||||
@@ -155,7 +175,7 @@ export async function appendHeartbeatRunEvent(
|
||||
? nativeSha256(input.nativeSource.canonicalPayload)
|
||||
: null;
|
||||
if (input.nativeSource) {
|
||||
const existing = await tx
|
||||
const existingRows = await tx
|
||||
.select()
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(
|
||||
@@ -168,20 +188,21 @@ export async function appendHeartbeatRunEvent(
|
||||
),
|
||||
),
|
||||
))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
.limit(2);
|
||||
const existing = existingRows[0];
|
||||
if (existing) {
|
||||
if (existing.sourcePayloadSha256 !== sourceHash) {
|
||||
if (existingRows.some(row => row.sourcePayloadSha256 !== sourceHash
|
||||
|| row.sourceInstanceId !== input.nativeSource!.sourceInstanceId
|
||||
|| row.sourceEventId !== input.nativeSource!.sourceEventId
|
||||
|| row.sourceSeq !== input.nativeSource!.sourceSeq)) {
|
||||
throw new HeartbeatRunEventConflictError();
|
||||
}
|
||||
if (!projectionCurrent) await requestNativeEventCursorRebuild(tx as unknown as Db, cursorRun!);
|
||||
return {
|
||||
row: existing,
|
||||
disposition: "duplicate" as const,
|
||||
highestContiguousSourceSeq: await contiguousCursor(
|
||||
tx as unknown as Db,
|
||||
input.runId,
|
||||
input.nativeSource.sourceInstanceId,
|
||||
),
|
||||
highestContiguousSourceSeq: await (projectionCurrent ? readNativeEventSourceCursor : readRetainedNativeEventCursor)(
|
||||
tx as unknown as Db, sourceScope!),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -196,9 +217,16 @@ export async function appendHeartbeatRunEvent(
|
||||
await notifyNativeResponsePresentation(tx, { companyId: run.companyId, runId: input.runId });
|
||||
}
|
||||
if (projectsRetention) await notifyAttentionRetentionWork(tx, { companyId: input.companyId, sourceKind: "failed_run" });
|
||||
let projectedSourceCursor = 0;
|
||||
if (input.nativeSource) {
|
||||
if (projectionCurrent) projectedSourceCursor = await projectNativeEventSourceRange(
|
||||
tx as unknown as Db, sourceScope!, input.nativeSource.sourceSeq);
|
||||
else await requestNativeEventCursorRebuild(tx as unknown as Db, cursorRun!);
|
||||
}
|
||||
const seq = await allocateHeartbeatRunEventSeq(
|
||||
tx as unknown as Db,
|
||||
input.runId,
|
||||
!input.nativeSource || projectionCurrent,
|
||||
);
|
||||
const [row] = await tx.insert(heartbeatRunEvents).values({
|
||||
companyId: input.companyId,
|
||||
@@ -222,47 +250,8 @@ export async function appendHeartbeatRunEvent(
|
||||
row,
|
||||
disposition: "committed" as const,
|
||||
highestContiguousSourceSeq: input.nativeSource
|
||||
? await contiguousCursor(
|
||||
tx as unknown as Db,
|
||||
input.runId,
|
||||
input.nativeSource.sourceInstanceId,
|
||||
)
|
||||
? projectionCurrent ? projectedSourceCursor : await readRetainedNativeEventCursor(tx as unknown as Db, sourceScope!)
|
||||
: 0,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
async function contiguousCursor(db: Db, runId: string, sourceInstanceId: string): Promise<number> {
|
||||
const scope = and(
|
||||
eq(heartbeatRunEvents.runId, runId),
|
||||
eq(heartbeatRunEvents.sourceInstanceId, sourceInstanceId),
|
||||
sql`${heartbeatRunEvents.sourceSeq} >= 1`,
|
||||
);
|
||||
const [range] = await db
|
||||
.select({
|
||||
count: sql<number>`count(*)`.mapWith(Number),
|
||||
max: sql<number | null>`max(${heartbeatRunEvents.sourceSeq})`.mapWith(Number),
|
||||
})
|
||||
.from(heartbeatRunEvents)
|
||||
.where(scope);
|
||||
if (!range || !range.count) return 0;
|
||||
if (range.max === range.count) return range.count;
|
||||
const [first] = await db
|
||||
.select({ sourceSeq: heartbeatRunEvents.sourceSeq })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(scope, eq(heartbeatRunEvents.sourceSeq, 1)))
|
||||
.limit(1);
|
||||
if (!first) return 0;
|
||||
const [last] = await db
|
||||
.select({ sourceSeq: heartbeatRunEvents.sourceSeq })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(scope, sql`not exists (
|
||||
select 1 from heartbeat_run_events next_event
|
||||
where next_event.run_id = heartbeat_run_events.run_id
|
||||
and next_event.source_instance_id = heartbeat_run_events.source_instance_id
|
||||
and next_event.source_seq = heartbeat_run_events.source_seq + 1
|
||||
)`))
|
||||
.orderBy(asc(heartbeatRunEvents.sourceSeq))
|
||||
.limit(1);
|
||||
return last?.sourceSeq ?? 0;
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
import { rebuildNativeEventCursorPage, type NativeEventCursorScope } from "./native-event-cursors.js";
|
||||
|
||||
type Entry = NativeEventCursorScope & { id: string; dueAt: number };
|
||||
|
||||
export function nativeEventCursorWork(db: Db) {
|
||||
async function load(scope?: NativeEventCursorScope, signal?: AbortSignal): Promise<Entry[]> {
|
||||
signal?.throwIfAborted();
|
||||
const rows = await db.select({ runId: heartbeatRuns.id, companyId: heartbeatRuns.companyId }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.eventCursorRebuildPending, true),
|
||||
scope ? and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.id, scope.runId)) : undefined));
|
||||
return rows.map(row => ({ ...row, id: row.runId, dueAt: 0 }));
|
||||
}
|
||||
async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
|
||||
const result = await rebuildNativeEventCursorPage(db, entry, signal);
|
||||
return result.pending ? { ...entry, dueAt: Date.now() + (result.contended ? 1000 : 0) } : null;
|
||||
}
|
||||
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.nativeEventCursorRebuild,
|
||||
minimumPassIntervalMs: 0,
|
||||
decodeScope: (hint: string): NativeEventCursorScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (entry: Entry, scope?: NativeEventCursorScope) => !scope ||
|
||||
entry.companyId === scope.companyId && entry.runId === scope.runId,
|
||||
load, check,
|
||||
}, canRun);
|
||||
}
|
||||
return { load, check, register };
|
||||
}
|
||||
@@ -0,0 +1,186 @@
|
||||
import { and, asc, eq, gt, lt, sql } from "drizzle-orm";
|
||||
import { heartbeatRunEventCursors, heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
||||
|
||||
export type NativeEventCursorScope = { companyId: string; runId: string };
|
||||
export type NativeEventCursorSource = NativeEventCursorScope & { sourceInstanceId: string };
|
||||
export type NativeEventCursorRun = Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"companyId" | "nextEventSeq" | "eventCursorProjectionSeq" | "eventCursorRebuildPending"> & { id: string };
|
||||
|
||||
export function nativeEventCursorProjectionCurrent(run: NativeEventCursorRun): boolean {
|
||||
return run.eventCursorProjectionSeq === run.nextEventSeq - 1;
|
||||
}
|
||||
|
||||
/** Caller holds the run lock; source state never grants execution authority. */
|
||||
export async function requestNativeEventCursorRebuild(db: Db, run: NativeEventCursorRun): Promise<void> {
|
||||
if (run.eventCursorRebuildPending) return;
|
||||
await notifyDeliveryWork(db, DELIVERY_QUEUES.nativeEventCursorRebuild,
|
||||
JSON.stringify({ companyId: run.companyId, runId: run.id }));
|
||||
await db.update(heartbeatRuns).set({ eventCursorRebuildPending: true })
|
||||
.where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId)));
|
||||
}
|
||||
|
||||
export async function readNativeEventSourceCursor(db: Db, scope: NativeEventCursorSource): Promise<number> {
|
||||
const [row] = await db.select({ cursor: heartbeatRunEventCursors.contiguousSourceSeq })
|
||||
.from(heartbeatRunEventCursors).where(and(
|
||||
eq(heartbeatRunEventCursors.companyId, scope.companyId),
|
||||
eq(heartbeatRunEventCursors.runId, scope.runId),
|
||||
eq(heartbeatRunEventCursors.sourceInstanceId, scope.sourceInstanceId),
|
||||
)).limit(1);
|
||||
return row?.cursor ?? 0;
|
||||
}
|
||||
|
||||
/** Incorporate one bounded received interval under the run lock. The ordinary
|
||||
* in-order path is one UPSERT regardless of the retained transcript length. */
|
||||
export async function projectNativeEventSourceRange(
|
||||
db: Db, scope: NativeEventCursorSource, startSeq: number, endSeq = startSeq,
|
||||
): Promise<number> {
|
||||
if (startSeq < 1 || endSeq < startSeq) return readNativeEventSourceCursor(db, scope);
|
||||
const rows = await db.execute(sql`
|
||||
with projected as (insert into heartbeat_run_event_cursors
|
||||
(company_id, run_id, source_instance_id, contiguous_source_seq, pending_range_count)
|
||||
values (${scope.companyId}, ${scope.runId}, ${scope.sourceInstanceId},
|
||||
case when ${startSeq}::bigint = 1 then ${endSeq}::bigint else 0 end, 0)
|
||||
on conflict (company_id, run_id, source_instance_id) do update
|
||||
set contiguous_source_seq = case
|
||||
when ${startSeq} <= heartbeat_run_event_cursors.contiguous_source_seq + 1
|
||||
then greatest(heartbeat_run_event_cursors.contiguous_source_seq, ${endSeq})
|
||||
else heartbeat_run_event_cursors.contiguous_source_seq end
|
||||
returning contiguous_source_seq as cursor, pending_range_count as pending),
|
||||
extended as (
|
||||
update heartbeat_run_event_source_ranges received
|
||||
set start_seq = least(received.start_seq, ${startSeq}),
|
||||
end_seq = greatest(received.end_seq, ${endSeq})
|
||||
from projected
|
||||
where projected.pending = 1 and projected.cursor < ${startSeq}::bigint - 1
|
||||
and received.company_id = ${scope.companyId} and received.run_id = ${scope.runId}
|
||||
and received.source_instance_id = ${scope.sourceInstanceId}
|
||||
and received.start_seq <= ${endSeq + 1} and received.end_seq >= ${startSeq - 1}
|
||||
returning received.start_seq
|
||||
)
|
||||
select projected.*, exists (select 1 from extended) as extended from projected
|
||||
`);
|
||||
let cursor = Number(rows[0].cursor), pending = Number(rows[0].pending);
|
||||
if (cursor >= endSeq && pending === 0 || rows[0].extended === true) return cursor;
|
||||
|
||||
if (cursor >= startSeq - 1) {
|
||||
// A newly closed hole may join a very large received suffix. Delete its
|
||||
// range metadata, never fetch the underlying historical event rows.
|
||||
const joined = await db.execute(sql`
|
||||
delete from heartbeat_run_event_source_ranges
|
||||
where company_id = ${scope.companyId} and run_id = ${scope.runId}
|
||||
and source_instance_id = ${scope.sourceInstanceId} and start_seq <= ${cursor + 1}
|
||||
returning end_seq
|
||||
`);
|
||||
if (joined.length === 0) return cursor;
|
||||
cursor = Math.max(cursor, ...joined.map(row => Number(row.end_seq)));
|
||||
pending -= joined.length;
|
||||
} else {
|
||||
// Only the predecessor and starts within this bounded input interval can
|
||||
// overlap or touch it. The primary key supports both indexed reads.
|
||||
const adjacent = await db.execute(sql`
|
||||
select start_seq, end_seq from (
|
||||
(select start_seq, end_seq from heartbeat_run_event_source_ranges
|
||||
where company_id = ${scope.companyId} and run_id = ${scope.runId}
|
||||
and source_instance_id = ${scope.sourceInstanceId} and start_seq <= ${startSeq}
|
||||
order by start_seq desc limit 1)
|
||||
union all
|
||||
(select start_seq, end_seq from heartbeat_run_event_source_ranges
|
||||
where company_id = ${scope.companyId} and run_id = ${scope.runId}
|
||||
and source_instance_id = ${scope.sourceInstanceId}
|
||||
and start_seq > ${startSeq} and start_seq <= ${endSeq + 1})
|
||||
) neighbors where end_seq >= ${startSeq - 1}
|
||||
`);
|
||||
const first = Math.min(startSeq, ...adjacent.map(row => Number(row.start_seq)));
|
||||
const last = Math.max(endSeq, ...adjacent.map(row => Number(row.end_seq)));
|
||||
if (adjacent.length > 0) await db.execute(sql`
|
||||
delete from heartbeat_run_event_source_ranges
|
||||
where company_id = ${scope.companyId} and run_id = ${scope.runId}
|
||||
and source_instance_id = ${scope.sourceInstanceId}
|
||||
and start_seq >= ${first} and start_seq <= ${endSeq + 1}
|
||||
`);
|
||||
await db.execute(sql`
|
||||
insert into heartbeat_run_event_source_ranges
|
||||
(company_id, run_id, source_instance_id, start_seq, end_seq)
|
||||
values (${scope.companyId}, ${scope.runId}, ${scope.sourceInstanceId}, ${first}, ${last})
|
||||
`);
|
||||
pending += 1 - adjacent.length;
|
||||
}
|
||||
await db.execute(sql`
|
||||
update heartbeat_run_event_cursors
|
||||
set contiguous_source_seq = ${cursor}, pending_range_count = ${pending}
|
||||
where company_id = ${scope.companyId} and run_id = ${scope.runId}
|
||||
and source_instance_id = ${scope.sourceInstanceId}
|
||||
`);
|
||||
return cursor;
|
||||
}
|
||||
|
||||
/** Correct compatibility path while an older writer's suffix is unprojected. */
|
||||
export async function readRetainedNativeEventCursor(db: Db, scope: NativeEventCursorSource): Promise<number> {
|
||||
const source = and(eq(heartbeatRunEvents.runId, scope.runId),
|
||||
eq(heartbeatRunEvents.sourceInstanceId, scope.sourceInstanceId), sql`${heartbeatRunEvents.sourceSeq} >= 1`);
|
||||
const [range] = await db.select({ count: sql<number>`count(*)`.mapWith(Number),
|
||||
max: sql<number | null>`max(${heartbeatRunEvents.sourceSeq})`.mapWith(Number) })
|
||||
.from(heartbeatRunEvents).where(source);
|
||||
if (!range?.count) return 0;
|
||||
if (range.max === range.count) return range.count;
|
||||
const [first] = await db.select({ sourceSeq: heartbeatRunEvents.sourceSeq }).from(heartbeatRunEvents)
|
||||
.where(and(source, eq(heartbeatRunEvents.sourceSeq, 1))).limit(1);
|
||||
if (!first) return 0;
|
||||
const [last] = await db.select({ sourceSeq: heartbeatRunEvents.sourceSeq }).from(heartbeatRunEvents)
|
||||
.where(and(source, sql`not exists (
|
||||
select 1 from heartbeat_run_events next_event
|
||||
where next_event.run_id = heartbeat_run_events.run_id
|
||||
and next_event.source_instance_id = heartbeat_run_events.source_instance_id
|
||||
and next_event.source_seq = heartbeat_run_events.source_seq + 1
|
||||
)`)).orderBy(asc(heartbeatRunEvents.sourceSeq)).limit(1);
|
||||
return last?.sourceSeq ?? 0;
|
||||
}
|
||||
|
||||
/** One bounded page per transaction. The marker detects older writers and
|
||||
* permits the projection to be discarded/rebuilt without changing receipts. */
|
||||
export async function rebuildNativeEventCursorPage(
|
||||
db: Db, scope: NativeEventCursorScope, signal?: AbortSignal,
|
||||
): Promise<{ pending: boolean; contended: boolean; processed: number }> {
|
||||
signal?.throwIfAborted();
|
||||
return db.transaction(async tx => {
|
||||
const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
|
||||
nextEventSeq: heartbeatRuns.nextEventSeq, eventCursorProjectionSeq: heartbeatRuns.eventCursorProjectionSeq,
|
||||
eventCursorRebuildPending: heartbeatRuns.eventCursorRebuildPending })
|
||||
.from(heartbeatRuns).where(and(eq(heartbeatRuns.id, scope.runId),
|
||||
eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.eventCursorRebuildPending, true)))
|
||||
.for("update", { skipLocked: true }).limit(1);
|
||||
if (!run) {
|
||||
const [pending] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.id, scope.runId), eq(heartbeatRuns.companyId, scope.companyId),
|
||||
eq(heartbeatRuns.eventCursorRebuildPending, true))).limit(1);
|
||||
return { pending: Boolean(pending), contended: Boolean(pending), processed: 0 };
|
||||
}
|
||||
const rows = await tx.select({ seq: heartbeatRunEvents.seq,
|
||||
sourceInstanceId: heartbeatRunEvents.sourceInstanceId, sourceSeq: heartbeatRunEvents.sourceSeq })
|
||||
.from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, run.id),
|
||||
gt(heartbeatRunEvents.seq, run.eventCursorProjectionSeq), lt(heartbeatRunEvents.seq, run.nextEventSeq)))
|
||||
.orderBy(asc(heartbeatRunEvents.seq)).limit(512);
|
||||
const sources = new Map<string, number[]>();
|
||||
for (const row of rows) if (row.sourceInstanceId !== null && row.sourceSeq !== null && row.sourceSeq >= 1) {
|
||||
const values = sources.get(row.sourceInstanceId) ?? [];
|
||||
values.push(row.sourceSeq); sources.set(row.sourceInstanceId, values);
|
||||
}
|
||||
for (const [sourceInstanceId, values] of sources) {
|
||||
values.sort((a, b) => a - b);
|
||||
let start = values[0], end = start;
|
||||
for (const value of values.slice(1)) {
|
||||
if (value <= end + 1) { end = Math.max(end, value); continue; }
|
||||
await projectNativeEventSourceRange(tx as unknown as Db, { ...scope, sourceInstanceId }, start, end);
|
||||
start = end = value;
|
||||
}
|
||||
await projectNativeEventSourceRange(tx as unknown as Db, { ...scope, sourceInstanceId }, start, end);
|
||||
}
|
||||
const through = rows.length < 512 ? run.nextEventSeq - 1 : rows.at(-1)!.seq;
|
||||
const pending = through < run.nextEventSeq - 1;
|
||||
signal?.throwIfAborted();
|
||||
await tx.update(heartbeatRuns).set({ eventCursorProjectionSeq: through, eventCursorRebuildPending: pending })
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
return { pending, contended: false, processed: rows.length };
|
||||
});
|
||||
}
|
||||
@@ -3624,6 +3624,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
const eventSeq = await allocateHeartbeatRunEventSeq(
|
||||
tx as unknown as Db,
|
||||
cancelled.id,
|
||||
true,
|
||||
);
|
||||
|
||||
await tx.insert(heartbeatRunEvents).values({
|
||||
|
||||
@@ -62,6 +62,7 @@ export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number })
|
||||
const workCheck = (reason: string, query: string | SQLWrapper) => ({ reason, query: typeof query === "string" ? sql.raw(query) : query });
|
||||
|
||||
const WORK_CHECKS = [
|
||||
workCheck("native_event_cursor_rebuild", `SELECT 1 FROM heartbeat_runs WHERE event_cursor_rebuild_pending = true`),
|
||||
workCheck("agents.execution_or_timer", `SELECT 1 FROM agents WHERE status = 'running' OR
|
||||
(status NOT IN ('terminated', 'paused', 'pending_approval') AND
|
||||
runtime_config->'heartbeat'->>'enabled' = 'true' AND
|
||||
|
||||
@@ -6,6 +6,8 @@ import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-
|
||||
import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js";
|
||||
import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { nativeEventCursorProjectionCurrent, projectNativeEventSourceRange, requestNativeEventCursorRebuild,
|
||||
} from "../heartbeat/native-event-cursors.js";
|
||||
|
||||
import { and, desc, eq, inArray, lt } from "drizzle-orm";
|
||||
|
||||
@@ -243,6 +245,7 @@ export class NativeRunCoordinatorStore {
|
||||
.limit(1);
|
||||
if (!run) throw new Error("native_event_run_not_authorized");
|
||||
|
||||
const projectionCurrent = nativeEventCursorProjectionCurrent(run);
|
||||
const [existing] = await tx
|
||||
.select()
|
||||
.from(heartbeatRunEvents)
|
||||
@@ -263,6 +266,7 @@ export class NativeRunCoordinatorStore {
|
||||
"source_event_replay_conflict",
|
||||
);
|
||||
}
|
||||
if (!projectionCurrent) await requestNativeEventCursorRebuild(tx as unknown as Db, run);
|
||||
const [latest] = await tx
|
||||
.select({ sourceSeq: heartbeatRunEvents.sourceSeq })
|
||||
.from(heartbeatRunEvents)
|
||||
@@ -303,6 +307,9 @@ export class NativeRunCoordinatorStore {
|
||||
if (isNativeResponseEvidence(event.eventType, canonicalPayload)) {
|
||||
await notifyNativeResponsePresentation(tx, this.#binding);
|
||||
}
|
||||
if (projectionCurrent) await projectNativeEventSourceRange(tx as unknown as Db,
|
||||
{ companyId: run.companyId, runId: run.id, sourceInstanceId: event.sourceInstanceId }, event.sourceSeq);
|
||||
else await requestNativeEventCursorRebuild(tx as unknown as Db, run);
|
||||
const cursor = run.nextEventSeq;
|
||||
const [inserted] = await tx
|
||||
.insert(heartbeatRunEvents)
|
||||
@@ -325,7 +332,8 @@ export class NativeRunCoordinatorStore {
|
||||
if (!inserted) throw new Error("native_event_not_persisted");
|
||||
await tx
|
||||
.update(heartbeatRuns)
|
||||
.set({ nextEventSeq: cursor + 1, updatedAt: new Date() })
|
||||
.set({ nextEventSeq: cursor + 1, updatedAt: new Date(),
|
||||
...(projectionCurrent ? { eventCursorProjectionSeq: cursor } : {}) })
|
||||
.where(eq(heartbeatRuns.id, this.#binding.runId));
|
||||
return {
|
||||
disposition: "committed" as const,
|
||||
|
||||
Reference in new issue
Block a user