| ... | @@ -272,3 +272,61 @@ it.effect("uses indexed high-water lookups for populated history without OR scan | ... | @@ -272,3 +272,61 @@ it.effect("uses indexed high-water lookups for populated history without OR scan |
| 272 | ), | 272 | ), |
| 273 | ), | 273 | ), |
| 274 | ); | 274 | ); |
| | 275 | |
| | 276 | it.effect("replays a command's events through the command index, not the sequence range", () => |
| | 277 | Effect.gen(function* () { |
| | 278 | const store = yield* OrchestrationEventStore.OrchestrationEventStore; |
| | 279 | const sql = yield* SqlClient.SqlClient; |
| | 280 | yield* runMigrations(); |
| | 281 | yield* sql` |
| | 282 | WITH RECURSIVE history(n) AS ( |
| | 283 | SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 25000 |
| | 284 | ) |
| | 285 | INSERT INTO orchestration_events ( |
| | 286 | event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, |
| | 287 | command_id, actor_kind, payload_json, metadata_json, application_event_version |
| | 288 | ) |
| | 289 | SELECT 'history:' || n, 'thread', 'thread', n, 'provider-session.detached', ${occurredAt}, |
| | 290 | CASE WHEN n IN (12, 24990) THEN 'retried-command' ELSE 'command:' || n END, |
| | 291 | 'server', |
| | 292 | '{"providerSessionId":"session","detachedAt":"' || ${occurredAt} || '"}', |
| | 293 | '{}', 2 |
| | 294 | FROM history |
| | 295 | `; |
| | 296 | const statements: Array<string> = []; |
| | 297 | const tracer = Tracer.make({ |
| | 298 | span(options) { |
| | 299 | const span = new Tracer.NativeSpan(options); |
| | 300 | const end = span.end.bind(span); |
| | 301 | span.end = (endTime, exit) => { |
| | 302 | end(endTime, exit); |
| | 303 | const query = span.attributes.get("db.query.text"); |
| | 304 | if (typeof query === "string") statements.push(query); |
| | 305 | }; |
| | 306 | return span; |
| | 307 | }, |
| | 308 | }); |
| | 309 | const replayed = yield* store |
| | 310 | .readAgentEvents({ commandId: CommandId.make("retried-command") }) |
| | 311 | .pipe(Stream.runCollect, Effect.withTracer(tracer)); |
| | 312 | assert.deepEqual( |
| | 313 | replayed.map((event) => event.sequence), |
| | 314 | [12, 24990], |
| | 315 | ); |
| | 316 | assert.equal(statements.length, 1); |
| | 317 | const plan = yield* sql.unsafe<{ readonly detail: string }>( |
| | 318 | `EXPLAIN QUERY PLAN ${statements[0]}`, |
| | 319 | [0, Number.MAX_SAFE_INTEGER, "retried-command", 500], |
| | 320 | ); |
| | 321 | assert.match( |
| | 322 | plan.map((row) => row.detail).join("\n"), |
| | 323 | /SEARCH orchestration_events USING INDEX idx_orch_events_command_id \(command_id=\?/, |
| | 324 | ); |
| | 325 | }).pipe( |
| | 326 | Effect.provide( |
| | 327 | OrchestrationEventStore.layer.pipe( |
| | 328 | Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })), |
| | 329 | ), |
| | 330 | ), |
| | 331 | ), |
| | 332 | ); |
Dara Adedeji <76637177+SunkenInTime@users.noreply.github.com> 2026-10-07 22:36:42-04:00
GitHub <noreply@github.com> 2026-10-07 19:36:42-07:00