diff --git a/.changeset/blob-offload.md b/.changeset/blob-offload.md new file mode 100644 index 0000000..8a46da6 --- /dev/null +++ b/.changeset/blob-offload.md @@ -0,0 +1,5 @@ +--- +'@swirls/looms': minor +--- + +`createLooms({ blobStore })` and `LoomsDurableObject` `blobs` store strings over 64 KiB by SHA-256 and keep a reference in the event or snapshot. A payload that is still over the 1 MiB event limit is stored as one blob. Reads, subscribers, folds, projections, and snapshot loads resolve the reference, so a run can complete with a payload larger than that limit. Trim removes a blob body once nothing retained still points at it. `memoryBlobStore`, `filesystemBlobStore`, and `r2BlobStore` are the included stores. Without a blob store, events over 1 MiB and snapshots over 8 MiB are still rejected. diff --git a/.changeset/effect-reliability.md b/.changeset/effect-reliability.md new file mode 100644 index 0000000..5e278e9 --- /dev/null +++ b/.changeset/effect-reliability.md @@ -0,0 +1,9 @@ +--- +'@swirls/looms': minor +--- + +Worker callbacks now answer `{ status: 'accepted' | 'stale' | 'cancelled', cancelRequested }` instead of the run state. `createLooms().workerCallback` and `createLoomsClient().workerCallback` return that result. `cancelRequested` tells an attempt that is already running to stop, because a dispatch that has left the host cannot be recalled. A stale report records nothing. + +`failEffect({ error, nonRetryable, retryAfterMs })` fails the current attempt. `nonRetryable` ends the effect even when the retry policy has attempts left. `retryAfterMs` waits at least that long before the next attempt and never shortens the policy backoff. Worker `fail` callbacks take the same two fields. + +A heartbeat may include `details`. The latest details are kept across retries and included on the next `EffectWorkerTask`, so a resumed attempt can continue from the recorded progress. diff --git a/.changeset/run-context.md b/.changeset/run-context.md new file mode 100644 index 0000000..d38fe7a --- /dev/null +++ b/.changeset/run-context.md @@ -0,0 +1,5 @@ +--- +'@swirls/looms': minor +--- + +Runs accept immutable `attributes` and a `secrets` context. Attributes are recorded on `runtime.run.started`, the run state, and the run summary, and are passed to `authorize` when present. Secrets stay in a side store: they are available to effect handlers and in-process worker dispatch, and they are not written to events, snapshots, or run responses. A worker whose transport persists tasks reads them with `POST /runs/:runId/effects/:effectId/:attempt/secrets` (`client.workerSecrets`), which only answers the attempt that is currently running. diff --git a/.oxlintrc.json b/.oxlintrc.json index e56ea0a..4c413cb 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -85,7 +85,8 @@ "files": [ "apps/demo/src/server.ts", "packages/debugger/src/server/**/*.ts", - "packages/s2/src/lite.ts" + "packages/s2/src/lite.ts", + "packages/runtime/src/filesystem-blob-store.ts" ], "rules": { "effecttsgo/global-fetch": "off", diff --git a/apps/docs/src/docs-manifest.json b/apps/docs/src/docs-manifest.json index 4b05ea8..1fdc569 100644 --- a/apps/docs/src/docs-manifest.json +++ b/apps/docs/src/docs-manifest.json @@ -830,7 +830,7 @@ "level": 2 } ], - "text": "Reliability and recovery Durability depends on the host and storage you choose. In-memory examples lose history on exit. SQLite persists locally; production actor hosts must preserve one writer per run and provide wake scheduling. Replay and recovery are different operations Historical replay folds recorded events to reconstruct state and does not dispatch effects. Recovery loads a snapshot and later events, then resumes unfinished work. An effect that lacks a recorded outcome may execute again. Looms does not promise exactly-once actions across arbitrary external services. A crash after an external action Looms requests an external action. The service performs it. The host stops before recording success. Recovery cannot infer success from the local log alone. Pass ctx.effectId as an idempotency key where the external API supports it. Check that the service retains the key for as long as you may need to retry the action. For unsupported APIs, store a durable operation ID, query the provider for the outcome, or reconcile before retrying. Never assume adding an arbitrary header makes an API idempotent. Retries and errors Effect retry policies count total attempts: maxAttempts: 3 means one initial attempt and up to two retries. Attempt starts and retry deadlines are recorded before the runtime waits, so a restart preserves the retry budget and backoff. Configure start, heartbeat, and execution timeouts for work delivered to external workers. Worker callbacks include the effect ID and attempt number; stale callbacks cannot complete a newer attempt. Exhaustion records runtime.effect.failed . Signals and webhook delivery Use the embed start/signal idempotencyKey option with a stable runId for retries of the same logical request. Verify webhook signatures and transform allowed webhook payloads into typed events on the server. Deduplicate by the provider delivery ID and correlate by run, thread, and domain identifiers. Retrying with a fresh key defeats deduplication. Interrupted model streams A model call can produce partial deltas before failing. Those deltas may already be recorded; resuming work may incur another model call and different output. Render partial content as provisional, inspect completion events, and decide how your application handles abandoned partial turns. The adapter does not guarantee exactly-once billing or identical regenerated text. Snapshots, retention, and projections A compatible snapshot bounds replay work; it is not a backup or a schema migration. Trimming events removes older audit and time-travel history and can prevent rebuilding new projections from the beginning. Define a retention and archive policy before enabling trimAfterSnapshot. Projectors run after local commit. Durable delivery stores a cursor per projector name and version, retries without advancing that cursor, and dead-letters exhausted batches. A run can finish before an external index or webhook catches up, so monitor lag and keep projector writes idempotent. Test recovery Test a restart while parked, an interruption during an external action, duplicate webhook delivery, and a deployment with existing runs. The testing guide provides the local restart test. Review versioning before changing definitions. ) }" + "text": "Reliability and recovery Durability depends on the host and storage you choose. In-memory examples lose history on exit. SQLite persists locally; production actor hosts must preserve one writer per run and provide wake scheduling. Replay and recovery are different operations Historical replay folds recorded events to reconstruct state and does not dispatch effects. Recovery loads a snapshot and later events, then resumes unfinished work. An effect that lacks a recorded outcome may execute again. Looms does not promise exactly-once actions across arbitrary external services. A crash after an external action Looms requests an external action. The service performs it. The host stops before recording success. Recovery cannot infer success from the local log alone. Pass ctx.effectId as an idempotency key where the external API supports it. Check that the service retains the key for as long as you may need to retry the action. For unsupported APIs, store a durable operation ID, query the provider for the outcome, or reconcile before retrying. Never assume adding an arbitrary header makes an API idempotent. Retries and errors Effect retry policies count total attempts: maxAttempts: 3 means one initial attempt and up to two retries. Attempt starts and retry deadlines are recorded before the runtime waits, so a restart preserves the retry budget and backoff. Configure start, heartbeat, and execution timeouts for work delivered to external workers. Worker callbacks include the effect ID and attempt number; stale callbacks cannot complete a newer attempt. Exhaustion records runtime.effect.failed . A handler can stop that budget early with failEffect and nonRetryable: true , or ask the next attempt to wait at least retryAfterMs without shortening the policy backoff. A worker heartbeat can carry details ; the next dispatch of that effect includes the latest details so the attempt can resume. The callback response is {'{ status, cancelRequested }'} , not the run state. cancelRequested tells an attempt that is already running to stop. Signals and webhook delivery Use the embed start/signal idempotencyKey option with a stable runId for retries of the same logical request. Verify webhook signatures and transform allowed webhook payloads into typed events on the server. Deduplicate by the provider delivery ID and correlate by run, thread, and domain identifiers. Retrying with a fresh key defeats deduplication. Interrupted model streams A model call can produce partial deltas before failing. Those deltas may already be recorded; resuming work may incur another model call and different output. Render partial content as provisional, inspect completion events, and decide how your application handles abandoned partial turns. The adapter does not guarantee exactly-once billing or identical regenerated text. Snapshots, retention, and projections A compatible snapshot bounds replay work; it is not a backup or a schema migration. Trimming events removes older audit and time-travel history and can prevent rebuilding new projections from the beginning. Define a retention and archive policy before enabling trimAfterSnapshot. Projectors run after local commit. Durable delivery stores a cursor per projector name and version, retries without advancing that cursor, and dead-letters exhausted batches. A run can finish before an external index or webhook catches up, so monitor lag and keep projector writes idempotent. Test recovery Test a restart while parked, an interruption during an external action, duplicate webhook delivery, and a deployment with existing runs. The testing guide provides the local restart test. Review versioning before changing definitions. ) }" }, { "path": "/docs/schedules", diff --git a/apps/docs/src/routes/docs.reliability.tsx b/apps/docs/src/routes/docs.reliability.tsx index 243e287..4ccc211 100644 --- a/apps/docs/src/routes/docs.reliability.tsx +++ b/apps/docs/src/routes/docs.reliability.tsx @@ -67,7 +67,13 @@ function Page() { a restart preserves the retry budget and backoff. Configure start, heartbeat, and execution timeouts for work delivered to external workers. Worker callbacks include the effect ID and attempt number; stale callbacks cannot complete a newer attempt. Exhaustion records{' '} - runtime.effect.failed. + runtime.effect.failed. A handler can stop that budget early with{' '} + failEffect and nonRetryable: true, or ask the next attempt to wait + at least retryAfterMs without shortening the policy backoff. A worker heartbeat + can carry details; the next dispatch of that effect includes the latest details + so the attempt can resume. The callback response is{' '} + {'{ status, cancelRequested }'}, not the run state.{' '} + cancelRequested tells an attempt that is already running to stop.

Signals and webhook delivery diff --git a/docs/production.md b/docs/production.md index 866dee0..a62ace9 100644 --- a/docs/production.md +++ b/docs/production.md @@ -17,15 +17,71 @@ All built-in stores reject an event over 1 MiB and a snapshot over 8 MiB, measured as serialized UTF-8, before storage is mutated. The stable error names are `EventPayloadTooLargeError` and `SnapshotPayloadTooLargeError`. +Pass `blobStore` to `createLooms`, or `blobs` to `LoomsDurableObject` / +`durableObjectEventStore`, to keep larger JSON. Strings over 64 KiB are replaced +with `{ "$looms": "blob", "sha256", "bytes", "mediaType": "application/json" }` +before those limits run. That object is reserved in stored JSON. If an event is +still over 1 MiB, its payload is stored as one blob. The body is SHA-256 +content, so identical JSON is stored once. `read`, live subscribers, folds, +projections, and snapshot loads resolve references, including references nested +inside another blob. Objects and arrays stay inline so snapshot state still +decodes; a snapshot that is over 8 MiB after large strings are removed is still +rejected. Trim deletes a body when no retained event and no latest snapshot in +any run still needs it. Without a blob store, oversized events and snapshots +are still rejected. + +`memoryBlobStore` is the in-memory implementation. `filesystemBlobStore(directory)` +is the Bun and Node implementation. `r2BlobStore(bucket)` writes an R2 bucket +under `looms/blobs/`. Every run's Durable Object can share one bucket. Each run +writes a marker per body it references, so trimming one run checks only that +body's markers, never other runs' logs. If a body is deleted while another run +is referencing it, that run's append fails and its retry uploads the body again. + +## Run attributes and secret context + +`attributes` on `start` are immutable labels (tenant, project, org). They are +written on `runtime.run.started`, folded into run state, and copied onto the +run summary. Authorization receives them on `route.attributes` when a run has +any. They must be safe for every reader of the run. + +`secrets` are bearer material for the run. The memory and SQLite stores keep +them in a side table, `looms_run_secrets`, which is not part of the event log +or a snapshot. Effect handlers read them on `ctx.secrets`. An in-process worker +receives them on the dispatch task. A transport that persists that task must +omit `secrets`. Its worker reads them with +`POST /runs/:runId/effects/:effectId/:attempt/secrets` (`client.workerSecrets`) +after its `started` callback is accepted. The route uses `worker` access, so it +returns 503 until the host configures worker auth. It answers +`{ status: 'accepted', secrets }` only while that attempt is started and +unsettled, and `{ status: 'stale' }` or `{ status: 'cancelled' }` otherwise. +Responses are sent with `cache-control: no-store`. A handler must not copy +secrets into an emitted event. + +A retry of the same start may omit `secrets` and keeps the stored context. Sending +a different context conflicts, and the error does not include the secret values. +Forks and effect-launched runs inherit the parent's attributes and secrets. +A store without `runSecrets` rejects a start that supplies secrets. + ## Worker effect protocol -1. The runtime dispatches `{ runId, effectId, attempt, type, input }`. +1. The runtime dispatches `{ runId, effectId, attempt, type, input, details? }`. + `details` is the progress carried by the previous attempt's last heartbeat. 2. The worker calls `started` before the schedule-to-start deadline. -3. Long work calls `heartbeat`; each accepted heartbeat extends the heartbeat - deadline without exceeding start-to-close. -4. The worker calls exactly one of `complete` or `fail`. +3. Long work calls `heartbeat`. Each accepted heartbeat extends the heartbeat + deadline without exceeding start-to-close. A heartbeat may include `details`; + the latest details are kept across retries and sent on the next dispatch. +4. The worker calls exactly one of `complete` or `fail`. `fail` may set + `nonRetryable` to stop the retry policy, or `retryAfterMs` to wait at least + that long before the next attempt (the policy backoff is never shortened). + A local handler does the same by failing with `failEffect`. 5. On cancellation, stop side effects and call `cancelled`. +Each callback response is `{ status: 'accepted' | 'stale' | 'cancelled', cancelRequested }`. +The host no longer returns the run state. `cancelRequested` is how a worker learns +that an attempt already in progress should stop: a queued dispatch cannot be recalled. +`stale` means the report was for another attempt, or the attempt had already settled, +and the host recorded nothing. + Callbacks include the same `effectId` and `attempt`. Stale callbacks are ignored. A dispatch transport failure is recorded as `ambiguous`: an operator must first verify the destination's idempotency record, then use `retryEffect` or diff --git a/packages/client/src/client.test.ts b/packages/client/src/client.test.ts index f454244..63ddbd5 100644 --- a/packages/client/src/client.test.ts +++ b/packages/client/src/client.test.ts @@ -66,6 +66,30 @@ describe('createLoomsClient', () => { ]) }) + test('workerSecrets posts to the attempt with the worker token', async () => { + const calls: { href: string; authorization?: string | null }[] = [] + + const client = createLoomsClient({ + workerCallbackToken: 'worker-token', + fetch: async (input, init) => { + calls.push({ + href: hrefOf(input), + authorization: new Headers(init?.headers).get('authorization'), + }) + + return Response.json({ status: 'accepted', secrets: { token: 'abc' } }) + }, + }) + + const result = await client.workerSecrets('run 1', { effectId: 'eff/1', attempt: 2 }) + + expect(result).toEqual({ status: 'accepted', secrets: { token: 'abc' } }) + + expect(calls).toEqual([ + { href: '/runs/run%201/effects/eff%2F1/2/secrets', authorization: 'Bearer worker-token' }, + ]) + }) + test('listDefinitions reads the definitions route', async () => { const calls: string[] = [] diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 413b98e..87d6efd 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -7,14 +7,20 @@ import { ReplayStepSchema, RunStateSchema, RunSummariesSchema, + SecretContextSchema, stringifyJson, type DefinitionInput, type DefinitionRef, type EventEnvelope, type EventInput, type JsonValue, + type RunAttributes, type RunState, + type SecretContext, + type WorkerCallback, type WorkerCallbackInput, + type WorkerCallbackResult, + type WorkerSecretsResult, } from '@looms/core' import { consumeSseStreamEffect, eventsFromSseData, sseFrames } from './sse' @@ -41,6 +47,62 @@ const StartResultSchema = Schema.Struct({ const RunResultSchema = Schema.Struct({ runId: Schema.String, state: RunStateSchema }) +interface WorkerFailRequestBody { + error: string + nonRetryable?: boolean + retryAfterMs?: number +} + +function attemptPath(runId: string, attempt: WorkerCallback): string { + return `/runs/${encodeURIComponent(runId)}/effects/${encodeURIComponent(attempt.effectId)}/${attempt.attempt}` +} + +function workerCallbackBody(input: WorkerCallbackInput) { + switch (input.kind) { + case 'complete': + return { events: input.events } + + case 'fail': { + const body: WorkerFailRequestBody = { error: input.error } + + if (input.nonRetryable !== undefined) { + body.nonRetryable = input.nonRetryable + } + + if (input.retryAfterMs !== undefined) { + body.retryAfterMs = input.retryAfterMs + } + + return body + } + + case 'heartbeat': + return input.details === undefined ? {} : { details: input.details } + case 'started': + case 'cancelled': + return {} + + default: { + const exhaustive: never = input + return exhaustive + } + } +} + +const WorkerCallbackResultSchema = Schema.Struct({ + status: Schema.Union([ + Schema.Literal('accepted'), + Schema.Literal('stale'), + Schema.Literal('cancelled'), + ]), + cancelRequested: Schema.Boolean, +}) + +const WorkerSecretsResultSchema = Schema.Union([ + Schema.Struct({ status: Schema.Literal('accepted'), secrets: SecretContextSchema }), + Schema.Struct({ status: Schema.Union([Schema.Literal('stale'), Schema.Literal('cancelled')]) }), +]) + /** `appended` counts the events the run gained while the wake ran. */ const WakeResultSchema = Schema.Struct({ runId: Schema.String, @@ -179,29 +241,35 @@ export function createLoomsClient(options: LoomsClientOptions = {}) { definitionVersion?: string input?: JsonValue runId?: string + attributes?: RunAttributes + secrets?: SecretContext }) => request('/runs', StartResultSchema, { method: 'POST', body: stringifyJson(args) }) - const workerCallback = (runId: string, input: WorkerCallbackInput) => { - const body = - input.kind === 'complete' - ? { events: input.events } - : input.kind === 'fail' - ? { error: input.error } - : {} - - return request( - `/runs/${encodeURIComponent(runId)}/effects/${encodeURIComponent(input.effectId)}/${input.attempt}/${input.kind}`, - RunResultSchema, - { - method: 'POST', - headers: options.workerCallbackToken - ? { authorization: `Bearer ${options.workerCallbackToken}` } - : undefined, - body: stringifyJson(body), - }, - ) + const workerHeaders = options.workerCallbackToken + ? { authorization: `Bearer ${options.workerCallbackToken}` } + : undefined + + const workerCallback = ( + runId: string, + input: WorkerCallbackInput, + ): Promise => { + const body = workerCallbackBody(input) + + return request(`${attemptPath(runId, input)}/${input.kind}`, WorkerCallbackResultSchema, { + method: 'POST', + headers: workerHeaders, + body: stringifyJson(body), + }) } + /** Read the run's secret context. Only the attempt that is currently running gets it. */ + const workerSecrets = (runId: string, attempt: WorkerCallback): Promise => + request(`${attemptPath(runId, attempt)}/secrets`, WorkerSecretsResultSchema, { + method: 'POST', + headers: workerHeaders, + body: '{}', + }) + const streamRun = (args: { kind: string definitionName: string @@ -300,6 +368,7 @@ export function createLoomsClient(options: LoomsClientOptions = {}) { body: stringifyJson(forkOptions?.runId === undefined ? {} : { runId: forkOptions.runId }), }), workerCallback, + workerSecrets, listRunSummaries: () => request('/runs/summaries', RunSummariesSchema), listDefinitions: () => request('/definitions', DefinitionCatalogSchema), replayTo: (runId: string, seq: number) => diff --git a/packages/cloudflare/src/durable-object.ts b/packages/cloudflare/src/durable-object.ts index d5746df..60047b7 100644 --- a/packages/cloudflare/src/durable-object.ts +++ b/packages/cloudflare/src/durable-object.ts @@ -6,6 +6,7 @@ import { createRemoteRunLauncher, type AnyRuntimeModule, type EventInput, + type BlobStore, type EventStoreTrimCoverage, type LaunchRunArgs, type RunLauncher, @@ -22,6 +23,8 @@ import type { DurableObjectNamespaceLike } from './router' export interface LoomsDurableObjectConfig { readonly modules: readonly AnyRuntimeModule[] readonly projectors?: readonly Projector[] + /** Content-addressed store for event and snapshot JSON over the inline byte limits. */ + readonly blobs?: BlobStore readonly snapshotEvery?: number readonly maxWakeIterations?: number readonly maxPendingRunOperations?: number @@ -87,6 +90,7 @@ export abstract class LoomsDurableObject extends DurableObject Effect.Effect + /** Offloads oversized event and snapshot JSON. See `withBlobOffload`. */ + readonly blobs?: BlobStore } export interface DurableObjectEventStore extends EventStore { @@ -32,7 +34,12 @@ export function durableObjectEventStore( transaction: (fn) => ctx.storage.transactionSync(fn), }) - const store = sqliteEventStore({ exec }) + const sqlite = sqliteEventStore({ exec }) + const snapshots = snapshotStoreOf(sqlite) + + const store = options.blobs + ? withBlobOffload(sqlite, options.blobs, snapshots ? { snapshots } : undefined) + : sqlite if (options.projectors && options.projectors.length > 0) { const delivery = createProjectorDelivery( diff --git a/packages/cloudflare/src/index.ts b/packages/cloudflare/src/index.ts index 47e4c3c..397f22f 100644 --- a/packages/cloudflare/src/index.ts +++ b/packages/cloudflare/src/index.ts @@ -11,6 +11,13 @@ export { type DurableObjectRunStub, type LoomsDurableObjectConfig, } from './durable-object' +export { + r2BlobStore, + type R2BucketLike, + type R2ListResult, + type R2ObjectBodyLike, + type R2StoredObject, +} from './r2-blob-store' export { routeToDurableObject, type DurableObjectNamespaceLike, diff --git a/packages/cloudflare/src/r2-blob-store.test.ts b/packages/cloudflare/src/r2-blob-store.test.ts new file mode 100644 index 0000000..b2b4207 --- /dev/null +++ b/packages/cloudflare/src/r2-blob-store.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, test } from 'bun:test' + +import { Effect } from 'effect' + +import { isBlobRef, isJsonObject, offloadJson, resolveJson } from '@looms/core' + +import { r2BlobStore, type R2BucketLike, type R2ListResult } from './r2-blob-store' + +function fakeBucket(): R2BucketLike & { readonly keys: () => string[] } { + const objects = new Map() + + return { + keys: () => [...objects.keys()], + + put: (key, value) => { + objects.set(key, value.slice()) + return Promise.resolve({ key }) + }, + + get: (key) => { + const body = objects.get(key) + + if (!body) { + return Promise.resolve(null) + } + + const copy = body.slice() + + return Promise.resolve({ + arrayBuffer: () => + Promise.resolve(copy.buffer.slice(copy.byteOffset, copy.byteOffset + copy.byteLength)), + }) + }, + + head: (key) => Promise.resolve(objects.has(key) ? { key } : null), + + delete: (key) => { + objects.delete(key) + return Promise.resolve(undefined) + }, + + list: (options) => { + const prefix = options?.prefix ?? '' + const keys = [...objects.keys()].filter((key) => key.startsWith(prefix)).toSorted() + const start = options?.cursor === undefined ? 0 : Number(options.cursor) + const page = keys.slice(start, start + 1) + const next = start + page.length + const truncated = next < keys.length + + const result: R2ListResult = { + objects: page.map((key) => ({ key })), + truncated, + } + + if (truncated) { + return Promise.resolve({ ...result, cursor: String(next) }) + } + + return Promise.resolve(result) + }, + } +} + +describe('r2BlobStore', () => { + test('pages the bucket and resolves an offloaded value', async () => { + const bucket = fakeBucket() + const blobs = r2BlobStore(bucket) + const value = { text: 'q'.repeat(80) } + const stored = await Effect.runPromise(offloadJson(value, blobs, { thresholdBytes: 16 })) + const text = isJsonObject(stored) ? stored.text : undefined + + if (text === undefined || !isBlobRef(text)) { + throw new Error('expected the string to be stored by hash') + } + + await Effect.runPromise(blobs.retain('run-a', [text.sha256])) + await Effect.runPromise(blobs.retain('run-b', [text.sha256])) + await Effect.runPromise(blobs.replaceRefs('run-a', new Set())) + + const resolved = await Effect.runPromise(resolveJson(stored, blobs)) + + expect(resolved).toEqual(value) + + await Effect.runPromise(blobs.replaceRefs('run-b', new Set())) + + expect(bucket.keys().filter((key) => key.includes(text.sha256))).toEqual([]) + }) +}) diff --git a/packages/cloudflare/src/r2-blob-store.ts b/packages/cloudflare/src/r2-blob-store.ts new file mode 100644 index 0000000..af6fb1f --- /dev/null +++ b/packages/cloudflare/src/r2-blob-store.ts @@ -0,0 +1,110 @@ +import { Effect } from 'effect' + +import { blobStoreFromBytes, BlobStoreError, type BlobBytes, type BlobStore } from '@looms/core' + +const PREFIX = 'looms/blobs/' + +export interface R2ObjectBodyLike { + arrayBuffer(): Promise +} + +export interface R2ListResult { + readonly objects: ReadonlyArray<{ readonly key: string }> + readonly truncated: boolean + readonly cursor?: string +} + +/** + * Structural subset of a Cloudflare `R2Bucket` binding. An `R2Bucket` satisfies it. + */ +export interface R2StoredObject { + readonly key: string +} + +export interface R2BucketLike { + put(key: string, value: Uint8Array): Promise + get(key: string): Promise + head(key: string): Promise + delete(key: string): Promise + list(options?: { prefix?: string; cursor?: string }): Promise +} + +/** + * Content-addressed blob store in R2 under `looms/blobs/`. Every run's Durable Object can + * share one bucket: each run writes its own reference markers, and a body is deleted only + * after no marker for it remains. + */ +export function r2BlobStore(bucket: R2BucketLike): BlobStore { + return blobStoreFromBytes(r2BlobBytes(bucket)) +} + +function r2BlobBytes(bucket: R2BucketLike): BlobBytes { + const keyFor = (key: string) => `${PREFIX}${key}` + + return { + read: (key) => + Effect.tryPromise({ + try: () => bucket.get(keyFor(key)), + catch: (cause) => new BlobStoreError(`Unable to read blob key ${key}`, cause), + }).pipe( + Effect.flatMap((object) => { + if (!object) { + return Effect.succeed(null) + } + + return Effect.tryPromise({ + try: () => object.arrayBuffer(), + catch: (cause) => new BlobStoreError(`Unable to read blob key ${key}`, cause), + }).pipe(Effect.map((buffer) => new Uint8Array(buffer))) + }), + ), + + write: (key, body) => + Effect.tryPromise({ + try: () => bucket.put(keyFor(key), body), + catch: (cause) => new BlobStoreError(`Unable to write blob key ${key}`, cause), + }).pipe(Effect.asVoid), + + remove: (key) => + Effect.tryPromise({ + try: () => bucket.delete(keyFor(key)), + catch: (cause) => new BlobStoreError(`Unable to delete blob key ${key}`, cause), + }).pipe(Effect.asVoid), + + exists: (key) => + Effect.tryPromise({ + try: () => bucket.head(keyFor(key)), + catch: (cause) => new BlobStoreError(`Unable to check blob key ${key}`, cause), + }).pipe(Effect.map((object) => object !== null)), + + list: (prefix) => listPrefixed(bucket, `${PREFIX}${prefix}`), + } +} + +function listPrefixed( + bucket: R2BucketLike, + prefix: string, +): Effect.Effect, BlobStoreError> { + return Effect.gen(function* () { + const keys: string[] = [] + let cursor: string | undefined + + for (;;) { + const page = yield* Effect.tryPromise({ + try: () => + cursor === undefined ? bucket.list({ prefix }) : bucket.list({ prefix, cursor }), + catch: (cause) => new BlobStoreError(`Unable to list blob keys ${prefix}`, cause), + }) + + for (const object of page.objects) { + keys.push(object.key.slice(PREFIX.length)) + } + + if (!page.truncated || page.cursor === undefined || page.cursor === cursor) { + return keys + } + + cursor = page.cursor + } + }) +} diff --git a/packages/core/src/blob-offload.test.ts b/packages/core/src/blob-offload.test.ts new file mode 100644 index 0000000..8a2090a --- /dev/null +++ b/packages/core/src/blob-offload.test.ts @@ -0,0 +1,257 @@ +import { describe, expect, test } from 'bun:test' + +import { Effect, Option, Stream } from 'effect' + +import { withBlobOffload } from './blob-offload' +import { BlobNotFoundError, memoryBlobStore } from './blob-store' +import { collectBlobDigests, expandBlobDigests, isBlobRef } from './blobs' +import { createEvent } from './envelope' +import { makeMemorySnapshotStore, snapshotStoreOf } from './snapshot-store' +import { emptyRunState } from './state' +import { makeMemoryEventStore } from './store' +import { isJsonObject, type JsonValue } from './types' + +function started(runId: string, payload: JsonValue) { + return createEvent(runId, { + type: 'runtime.run.started', + payload, + threadId: null, + origin: { type: 'system' }, + }) +} + +function firstDigest(value: JsonValue): string | undefined { + if (isBlobRef(value)) { + return value.sha256 + } + + if (Array.isArray(value)) { + for (const item of value) { + const found = firstDigest(item) + + if (found) { + return found + } + } + + return undefined + } + + if (isJsonObject(value)) { + for (const child of Object.values(value)) { + const found = firstDigest(child) + + if (found) { + return found + } + } + } + + return undefined +} + +describe('blob offload', () => { + test('keeps small JSON inline and resolves values over the threshold', async () => { + const inner = await Effect.runPromise(makeMemoryEventStore) + const blobs = memoryBlobStore() + const store = withBlobOffload(inner, blobs, { thresholdBytes: 32 }) + const large = 'x'.repeat(80) + + await Effect.runPromise(store.append('run_small', [started('run_small', { n: 1 })])) + + await Effect.runPromise(store.append('run_large', [started('run_large', { blob: large })])) + + const small = await Effect.runPromise(inner.read('run_small')) + expect(small[0]?.payload).toEqual({ n: 1 }) + expect(JSON.stringify(small)).not.toContain('"$looms"') + + const raw = await Effect.runPromise(inner.read('run_large')) + expect(JSON.stringify(raw)).toContain('"$looms":"blob"') + expect(JSON.stringify(raw)).not.toContain(large) + + const resolved = await Effect.runPromise(store.read('run_large')) + expect(resolved[0]?.payload).toEqual({ blob: large }) + + const live = await Effect.runPromise( + store.subscribe('run_large', { fromSeq: 1 }).pipe(Stream.take(1), Stream.runCollect), + ) + + expect(live[0]?.payload).toEqual({ blob: large }) + }) + + test('stores one body for the same bytes and deletes it only after every run releases it', async () => { + const inner = await Effect.runPromise(makeMemoryEventStore) + const blobs = memoryBlobStore() + const store = withBlobOffload(inner, blobs, { thresholdBytes: 32 }) + const payload = { blob: 'x'.repeat(80) } + + await Effect.runPromise(store.append('run_a', [started('run_a', payload)])) + await Effect.runPromise(store.append('run_b', [started('run_b', payload)])) + + const raw = await Effect.runPromise(inner.read('run_a')) + const digest = firstDigest(raw[0]?.payload ?? null) + + expect(digest).toBeDefined() + + await Effect.runPromise(store.trim!('run_a', 10)) + + const surviving = await Effect.runPromise(store.read('run_b')) + expect(surviving[0]?.payload).toEqual(payload) + + await Effect.runPromise(store.trim!('run_b', 10)) + + const missing = await Effect.runPromise( + blobs.get(digest!).pipe( + Effect.match({ + onFailure: (error) => error, + onSuccess: () => null, + }), + ), + ) + + expect(missing).toBeInstanceOf(BlobNotFoundError) + }) + + test('deletes nested bodies when the parent reference is trimmed', async () => { + const inner = await Effect.runPromise(makeMemoryEventStore) + const blobs = memoryBlobStore() + const store = withBlobOffload(inner, blobs, { thresholdBytes: 32, maxEventBytes: 200 }) + const text = 'y'.repeat(80) + const payload = [text, text, text, text] + + await Effect.runPromise(store.append('run_nested', [started('run_nested', payload)])) + + const raw = await Effect.runPromise(inner.read('run_nested')) + const seeds = new Set() + collectBlobDigests(raw[0]?.payload ?? null, seeds) + + const expanded = await Effect.runPromise(expandBlobDigests(blobs, seeds)) + expect(expanded.size).toBeGreaterThan(1) + expect(isBlobRef(raw[0]?.payload ?? null)).toBe(true) + + const resolved = await Effect.runPromise(store.read('run_nested')) + expect(resolved[0]?.payload).toEqual(payload) + + await Effect.runPromise(store.trim!('run_nested', 10)) + + for (const digest of expanded) { + const missing = await Effect.runPromise( + blobs.get(digest).pipe( + Effect.match({ + onFailure: (error) => error, + onSuccess: () => null, + }), + ), + ) + + expect(missing).toBeInstanceOf(BlobNotFoundError) + } + }) + + test('fails a read that names a missing blob and ignores a lookalike object', async () => { + const inner = await Effect.runPromise(makeMemoryEventStore) + const blobs = memoryBlobStore() + const store = withBlobOffload(inner, blobs, { thresholdBytes: 32 }) + + const lookalike = { + $looms: 'blob', + sha256: 'b'.repeat(64), + bytes: 2, + mediaType: 'application/json', + extra: true, + } + + await Effect.runPromise( + inner.append('run_missing', [ + started('run_missing', { + $looms: 'blob', + sha256: 'a'.repeat(64), + bytes: 2, + mediaType: 'application/json', + }), + ]), + ) + + await Effect.runPromise(store.append('run_lookalike', [started('run_lookalike', lookalike)])) + + const missing = await Effect.runPromise( + store.read('run_missing').pipe( + Effect.match({ + onFailure: (error) => error, + onSuccess: () => null, + }), + ), + ) + + expect(missing).toBeInstanceOf(Error) + expect(String(missing)).toContain('was not found') + + const kept = await Effect.runPromise(store.read('run_lookalike')) + expect(kept[0]?.payload).toEqual(lookalike) + }) + + test('offloads snapshot state and keeps those blobs when the log is trimmed', async () => { + const inner = await Effect.runPromise(makeMemoryEventStore) + const snapshots = await Effect.runPromise(makeMemorySnapshotStore) + const blobs = memoryBlobStore() + const store = withBlobOffload(inner, blobs, { snapshots, thresholdBytes: 32 }) + const saved = snapshotStoreOf(store) + const large = 'z'.repeat(200) + + const state = { + ...emptyRunState('run_snap'), + startIdentity: { + kind: 'workflow', + definitionName: 'blob', + definitionVersion: 'v1', + input: { blob: large }, + requestedThreadId: null, + idempotencyKey: null, + rootThreadId: 'thr_root', + launchedBy: null, + correlationId: null, + }, + } + + expect(saved).toBeDefined() + + await Effect.runPromise( + saved!.save({ + runId: 'run_snap', + cursor: 1, + stateHash: 'hash', + takenAt: 1, + state, + }), + ) + + const raw = await Effect.runPromise(snapshots.loadLatest('run_snap')) + expect(Option.isSome(raw)).toBe(true) + + if (Option.isSome(raw)) { + const encoded = JSON.stringify(raw.value.state) + expect(encoded).toContain('"$looms":"blob"') + expect(encoded).not.toContain(large) + } + + const loaded = await Effect.runPromise(saved!.loadLatest('run_snap')) + expect(Option.isSome(loaded)).toBe(true) + + if (Option.isSome(loaded)) { + expect(loaded.value.state.startIdentity?.input).toEqual({ blob: large }) + } + + await Effect.runPromise( + store.append('run_snap', [started('run_snap', { other: 'q'.repeat(80) })]), + ) + + await Effect.runPromise(store.trim!('run_snap', 10)) + + const afterTrim = await Effect.runPromise(saved!.loadLatest('run_snap')) + expect(Option.isSome(afterTrim)).toBe(true) + + if (Option.isSome(afterTrim)) { + expect(afterTrim.value.state.startIdentity?.input).toEqual({ blob: large }) + } + }) +}) diff --git a/packages/core/src/blob-offload.ts b/packages/core/src/blob-offload.ts new file mode 100644 index 0000000..680da6b --- /dev/null +++ b/packages/core/src/blob-offload.ts @@ -0,0 +1,282 @@ +import { Effect, Option, Schema, Stream } from 'effect' + +import { BlobStoreError, type BlobStore, type BlobStoreFailure } from './blob-store' +import { + DEFAULT_BLOB_THRESHOLD_BYTES, + claimJson, + collectBlobDigests, + expandBlobDigests, + offloadJson, + resolveJson, +} from './blobs' +import type { AppendableEvent, EventEnvelope } from './envelope' +import { DEFAULT_MAX_EVENT_BYTES, utf8JsonBytes } from './limits' +import { SnapshotStoreError, withSnapshotStore, type SnapshotStore } from './snapshot-store' +import { RunStateSchema, type RunState } from './state' +import { EventStoreError, type EventStore } from './store' +import type { JsonValue } from './types' + +const decodeJsonText = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Json)) +const decodeRunState = Schema.decodeUnknownEffect(RunStateSchema) + +export interface BlobOffloadOptions { + /** Snapshot store to offload and resolve. Trim keeps digests referenced by its latest snapshot. */ + readonly snapshots?: SnapshotStore + /** UTF-8 JSON size above which a string is stored by hash. Default 64 KiB. */ + readonly thresholdBytes?: number + /** Serialized event size that claims the whole payload. Default 1 MiB. */ + readonly maxEventBytes?: number +} + +/** + * Store oversized JSON outside the event log. + * + * Append replaces values over the threshold with a content-addressed reference before the + * event byte limit runs. `read`, `readStream`, and `subscribe` resolve those references, so + * folds, projections, and callers see the original values. The stored log keeps the references. + * When `snapshots` is set, snapshot save and load do the same, and trim deletes a body only + * once no retained event and no latest snapshot in any run still needs it. + */ +export function withBlobOffload( + store: EventStore, + blobs: BlobStore, + options?: BlobOffloadOptions, +): EventStore { + const thresholdBytes = options?.thresholdBytes ?? DEFAULT_BLOB_THRESHOLD_BYTES + const maxEventBytes = options?.maxEventBytes ?? DEFAULT_MAX_EVENT_BYTES + const snapshots = options?.snapshots + + const wrappedSnapshots = snapshots + ? wrapSnapshots(snapshots, store, blobs, thresholdBytes) + : undefined + + const release = (runId: string): Effect.Effect => + keepDigests(store, snapshots, blobs, runId).pipe( + Effect.flatMap((keep) => blobs.replaceRefs(runId, keep)), + Effect.mapError(asEventError), + ) + + const wrapped: EventStore = { + append: (runId, events, appendOptions) => + Effect.gen(function* () { + const stored = yield* Effect.forEach(events, (event) => + offloadAppendable(runId, event, blobs, thresholdBytes, maxEventBytes), + ).pipe(Effect.mapError(asEventError)) + + const seeds = digestsOf(stored.map((event) => event.payload)) + + const expanded = yield* expandBlobDigests(blobs, seeds).pipe(Effect.mapError(asEventError)) + + yield* blobs.retain(runId, [...expanded]).pipe(Effect.mapError(asEventError)) + + return yield* store.append(runId, stored, appendOptions) + }), + + read: (runId, readOptions) => + store + .read(runId, readOptions) + .pipe( + Effect.flatMap((events) => + Effect.forEach(events, (event) => + resolveEnvelope(event, blobs).pipe(Effect.mapError(asEventError)), + ), + ), + ), + + readStream: store.readStream + ? (runId, readOptions) => + store.readStream!(runId, readOptions).pipe( + Stream.mapEffect((event) => + resolveEnvelope(event, blobs).pipe(Effect.mapError(asEventError)), + ), + ) + : undefined, + + tail: (runId) => store.tail(runId), + + bounds: store.bounds ? (runId) => store.bounds!(runId) : undefined, + + trim: store.trim + ? (runId, beforeSeq) => store.trim!(runId, beforeSeq).pipe(Effect.andThen(release(runId))) + : undefined, + + subscribe: (runId, subscribeOptions) => + store + .subscribe(runId, subscribeOptions) + .pipe( + Stream.mapEffect((event) => + resolveEnvelope(event, blobs).pipe(Effect.mapError(asEventError)), + ), + ), + + listRuns: store.listRuns, + } + + if (!wrappedSnapshots) { + return wrapped + } + + return withSnapshotStore(wrapped, wrappedSnapshots) +} + +function offloadAppendable( + runId: string, + event: AppendableEvent, + blobs: BlobStore, + thresholdBytes: number, + maxEventBytes: number, +): Effect.Effect { + return Effect.gen(function* () { + const payload = yield* offloadJson(event.payload, blobs, { thresholdBytes }) + const offloaded = { ...event, payload } + + if (eventBytes(runId, offloaded) <= maxEventBytes) { + return offloaded + } + + const ref = yield* claimJson(payload, blobs) + + return { ...event, payload: ref } + }) +} + +function eventBytes(runId: string, event: AppendableEvent): number { + return utf8JsonBytes({ ...event, runId, seq: event.seq ?? 0 }) +} + +function resolveEnvelope( + event: EventEnvelope, + blobs: BlobStore, +): Effect.Effect { + return resolveJson(event.payload, blobs).pipe(Effect.map((payload) => ({ ...event, payload }))) +} + +function digestsOf(values: ReadonlyArray): Set { + const seeds = new Set() + + for (const value of values) { + collectBlobDigests(value, seeds) + } + + return seeds +} + +function keepDigests( + store: EventStore, + snapshots: SnapshotStore | undefined, + blobs: BlobStore, + runId: string, +): Effect.Effect, EventStoreError | BlobStoreFailure> { + return Effect.gen(function* () { + const events = yield* store.read(runId) + const seeds = digestsOf(events.map((event) => event.payload)) + + if (snapshots) { + const latest = yield* snapshots + .loadLatest(runId) + .pipe(Effect.mapError((cause) => new EventStoreError(cause.message, cause))) + + if (Option.isSome(latest)) { + const encoded = yield* jsonValue(latest.value.state).pipe( + Effect.mapError((cause) => new EventStoreError(cause.message, cause)), + ) + + collectBlobDigests(encoded, seeds) + } + } + + return yield* expandBlobDigests(blobs, seeds) + }) +} + +function wrapSnapshots( + snapshots: SnapshotStore, + store: EventStore, + blobs: BlobStore, + thresholdBytes: number, +): SnapshotStore { + const release = (runId: string): Effect.Effect => + keepDigests(store, snapshots, blobs, runId).pipe( + Effect.flatMap((keep) => blobs.replaceRefs(runId, keep)), + Effect.mapError((cause) => new SnapshotStoreError(cause.message, cause)), + ) + + return { + loadLatest: (runId) => + Effect.gen(function* () { + const loaded = yield* snapshots.loadLatest(runId) + + if (Option.isNone(loaded)) { + return loaded + } + + const encoded = yield* jsonValue(loaded.value.state).pipe(Effect.mapError(asSnapshotError)) + const resolved = yield* resolveJson(encoded, blobs).pipe(Effect.mapError(asSnapshotError)) + + const state = yield* decodeRunState(resolved).pipe( + Effect.mapError( + (cause) => new SnapshotStoreError('snapshot state did not match RunState', cause), + ), + ) + + return Option.some({ ...loaded.value, state }) + }), + + save: (snapshot) => + Effect.gen(function* () { + const encoded = yield* jsonValue(snapshot.state).pipe(Effect.mapError(asSnapshotError)) + + const offloaded = yield* offloadJson(encoded, blobs, { + thresholdBytes, + claimRoot: false, + }).pipe(Effect.mapError(asSnapshotError)) + + const state = yield* decodeRunState(offloaded).pipe( + Effect.mapError( + (cause) => new SnapshotStoreError('snapshot state did not match RunState', cause), + ), + ) + + const seeds = digestsOf([offloaded]) + + const expanded = yield* expandBlobDigests(blobs, seeds).pipe( + Effect.mapError(asSnapshotError), + ) + + yield* blobs.retain(snapshot.runId, [...expanded]).pipe(Effect.mapError(asSnapshotError)) + + return yield* snapshots.save({ + runId: snapshot.runId, + cursor: snapshot.cursor, + stateHash: snapshot.stateHash, + takenAt: snapshot.takenAt, + state, + }) + }), + + listCursors: snapshots.listCursors ? (runId) => snapshots.listCursors!(runId) : undefined, + + prune: snapshots.prune + ? (runId, keepLatest) => + snapshots.prune!(runId, keepLatest).pipe(Effect.andThen(release(runId))) + : undefined, + + saveHeader: (header) => snapshots.saveHeader(header), + + listHeaders: snapshots.listHeaders, + } +} + +function jsonValue(state: RunState): Effect.Effect { + return decodeJsonText(JSON.stringify(state)).pipe( + Effect.mapError((cause) => new BlobStoreError('snapshot state is not JSON', cause)), + ) +} + +function asEventError(cause: BlobStoreFailure | EventStoreError): EventStoreError { + return new EventStoreError(cause.message, cause) +} + +function asSnapshotError(cause: BlobStoreFailure | EventStoreError): SnapshotStoreError { + return new SnapshotStoreError(cause.message, cause) +} diff --git a/packages/core/src/blob-store.test.ts b/packages/core/src/blob-store.test.ts new file mode 100644 index 0000000..f18eedd --- /dev/null +++ b/packages/core/src/blob-store.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, test } from 'bun:test' + +import { Effect } from 'effect' + +import { blobStoreFromBytes, sha256Hex, type BlobBytes } from './blob-store' + +interface Recorded { + readonly bytes: BlobBytes + readonly objects: Map + readonly reads: string[] + onRemove?: (key: string) => void +} + +function recordedBytes(): Recorded { + const objects = new Map() + const reads: string[] = [] + + const recorded: Recorded = { + objects, + reads, + bytes: { + read: (key) => + Effect.sync(() => { + reads.push(key) + return objects.get(key) ?? null + }), + write: (key, body) => + Effect.sync(() => { + objects.set(key, body) + }), + remove: (key) => + Effect.sync(() => { + objects.delete(key) + recorded.onRemove?.(key) + }), + exists: (key) => Effect.sync(() => objects.has(key)), + list: (prefix) => + Effect.sync(() => [...objects.keys()].filter((key) => key.startsWith(prefix))), + }, + } + + return recorded +} + +const body = new TextEncoder().encode('"a large string"') +const digest = await Effect.runPromise(sha256Hex(body)) +const objectKey = `objects/${digest}` + +describe('blob references', () => { + test('a shared body survives until the last run drops it', async () => { + const recorded = recordedBytes() + const blobs = blobStoreFromBytes(recorded.bytes) + + await Effect.runPromise(blobs.put(digest, body)) + await Effect.runPromise(blobs.retain('run-a', [digest])) + await Effect.runPromise(blobs.retain('run-b', [digest])) + await Effect.runPromise(blobs.replaceRefs('run-a', new Set())) + + expect(recorded.objects.has(objectKey)).toBe(true) + + await Effect.runPromise(blobs.replaceRefs('run-b', new Set())) + + expect([...recorded.objects.keys()].filter((key) => key.includes(digest))).toEqual([]) + }) + + test('dropping a reference never reads another run', async () => { + const recorded = recordedBytes() + const blobs = blobStoreFromBytes(recorded.bytes) + + await Effect.runPromise(blobs.put(digest, body)) + + for (const runId of ['run-a', 'run-b', 'run-c', 'run-d']) { + await Effect.runPromise(blobs.retain(runId, [digest])) + } + + recorded.reads.length = 0 + await Effect.runPromise(blobs.replaceRefs('run-a', new Set())) + + expect(recorded.reads).toEqual(['runs/run-a']) + }) + + test('a reference recorded after the body was deleted fails so the append can retry', async () => { + const recorded = recordedBytes() + const blobs = blobStoreFromBytes(recorded.bytes) + + await Effect.runPromise(blobs.put(digest, body)) + await Effect.runPromise(blobs.retain('run-a', [digest])) + await Effect.runPromise(blobs.put(digest, body)) + await Effect.runPromise(blobs.replaceRefs('run-a', new Set())) + + const failed = await Effect.runPromise(Effect.flip(blobs.retain('run-b', [digest]))) + + expect(failed.message).toContain('retry the append') + + await Effect.runPromise(blobs.put(digest, body)) + await Effect.runPromise(blobs.retain('run-b', [digest])) + + expect(recorded.objects.has(objectKey)).toBe(true) + }) + + test('a body deleted while another run references it is put back', async () => { + const recorded = recordedBytes() + const blobs = blobStoreFromBytes(recorded.bytes) + + await Effect.runPromise(blobs.put(digest, body)) + await Effect.runPromise(blobs.retain('run-a', [digest])) + + recorded.onRemove = (key) => { + if (key === objectKey) { + recorded.objects.set(`refs/${digest}/run-b`, new Uint8Array(0)) + } + } + + await Effect.runPromise(blobs.replaceRefs('run-a', new Set())) + + expect(recorded.objects.get(objectKey)).toEqual(body) + }) +}) diff --git a/packages/core/src/blob-store.ts b/packages/core/src/blob-store.ts new file mode 100644 index 0000000..f9c54ec --- /dev/null +++ b/packages/core/src/blob-store.ts @@ -0,0 +1,294 @@ +import { Data, Effect, Schema } from 'effect' + +const SHA256 = /^[a-f0-9]{64}$/ + +const decodeDigestList = Schema.decodeUnknownEffect( + Schema.fromJsonString(Schema.Array(Schema.String)), +) + +export class BlobStoreError extends Data.TaggedError('BlobStoreError')<{ + readonly message: string + readonly cause?: unknown +}> { + override readonly cause?: unknown + + constructor(message: string, cause?: unknown) { + super({ message, cause }) + this.name = 'BlobStoreError' + this.cause = cause + } +} + +export class BlobNotFoundError extends Data.TaggedError('BlobNotFoundError')<{ + readonly digest: string + readonly message: string +}> { + constructor(digest: string) { + super({ digest, message: `blob ${digest} was not found` }) + this.name = 'BlobNotFoundError' + } +} + +export type BlobStoreFailure = BlobStoreError | BlobNotFoundError + +/** + * Content-addressed bytes. `put` is idempotent: the same digest stores one body. + * `retain` and `replaceRefs` record which digests a run still needs. A body is deleted only + * when no run references it. Many runs may share one store, each writing its own references. + */ +export interface BlobStore { + readonly put: (digest: string, body: Uint8Array) => Effect.Effect + readonly get: (digest: string) => Effect.Effect + readonly retain: ( + runId: string, + digests: ReadonlyArray, + ) => Effect.Effect + readonly replaceRefs: ( + runId: string, + keep: ReadonlySet, + ) => Effect.Effect +} + +export function isSha256Digest(value: string): boolean { + return SHA256.test(value) +} + +/** Byte operations behind {@link blobStoreFromBytes}. Keys are store-relative, not run ids. */ +export interface BlobBytes { + readonly read: (key: string) => Effect.Effect + readonly write: (key: string, body: Uint8Array) => Effect.Effect + readonly remove: (key: string) => Effect.Effect + readonly exists: (key: string) => Effect.Effect + readonly list: (prefix: string) => Effect.Effect, BlobStoreError> +} + +const EMPTY = new Uint8Array(0) + +export function memoryBlobStore(): BlobStore { + return blobStoreFromBytes(memoryBlobBytes()) +} + +export function blobStoreFromBytes(bytes: BlobBytes): BlobStore { + return { + put: (digest, body) => + Effect.gen(function* () { + yield* assertDigest(digest) + + const actual = yield* sha256Hex(body) + + if (actual !== digest) { + return yield* new BlobStoreError(`blob body does not match digest ${digest}`) + } + + const existing = yield* bytes.read(objectKey(digest)) + + if (existing) { + return yield* Effect.void + } + + return yield* bytes.write(objectKey(digest), body) + }), + + get: (digest) => + Effect.gen(function* () { + yield* assertDigest(digest) + + const body = yield* bytes.read(objectKey(digest)) + + if (!body) { + return yield* new BlobNotFoundError(digest) + } + + const actual = yield* sha256Hex(body) + + if (actual !== digest) { + return yield* new BlobStoreError(`blob ${digest} failed its integrity check`) + } + + return body + }), + + // Markers go down before the existence check, and `collect` lists markers again after + // deleting. Whichever side runs second sees the other, so a referenced body is never lost: + // either `collect` restores it or this append fails and its retry uploads the body again. + retain: (runId, digests) => + Effect.gen(function* () { + const current = new Set(yield* readRunDigests(bytes, runId)) + const added = [...new Set(digests)].filter((digest) => !current.has(digest)) + + if (added.length === 0) { + return yield* Effect.void + } + + yield* Effect.forEach(added, assertDigest, { discard: true }) + + yield* Effect.forEach(added, (digest) => bytes.write(refKey(digest, runId), EMPTY), { + concurrency: REF_CONCURRENCY, + discard: true, + }) + + yield* writeRunDigests(bytes, runId, [...current, ...added]) + + const present = yield* Effect.forEach(added, (digest) => bytes.exists(objectKey(digest)), { + concurrency: REF_CONCURRENCY, + }) + + const missing = added.filter((_, index) => !present[index]) + + if (missing.length > 0) { + return yield* new BlobStoreError( + `blobs ${missing.join(', ')} were deleted before run ${runId} referenced them; retry the append`, + ) + } + + return yield* Effect.void + }), + + replaceRefs: (runId, keep) => + Effect.gen(function* () { + const previous = yield* readRunDigests(bytes, runId) + const dropped = previous.filter((digest) => !keep.has(digest)) + + if (dropped.length === 0) { + return + } + + yield* Effect.forEach( + dropped, + (digest) => + bytes.remove(refKey(digest, runId)).pipe(Effect.andThen(collect(bytes, digest))), + { concurrency: REF_CONCURRENCY, discard: true }, + ) + + yield* writeRunDigests( + bytes, + runId, + previous.filter((digest) => keep.has(digest)), + ) + }), + } +} + +const REF_CONCURRENCY = 8 + +/** Deletes a body no run references, and puts it back if a run referenced it meanwhile. */ +function collect(bytes: BlobBytes, digest: string): Effect.Effect { + const referenced = bytes.list(refPrefix(digest)).pipe(Effect.map((keys) => keys.length > 0)) + + return Effect.gen(function* () { + if (yield* referenced) { + return + } + + const body = yield* bytes.read(objectKey(digest)) + + if (!body) { + return + } + + yield* bytes.remove(objectKey(digest)) + + if (yield* referenced) { + yield* bytes.write(objectKey(digest), body) + } + }) +} + +function memoryBlobBytes(): BlobBytes { + const objects = new Map() + + return { + read: (key) => + Effect.sync(() => { + const body = objects.get(key) + + return body ? body.slice() : null + }), + + write: (key, body) => + Effect.sync(() => { + objects.set(key, body.slice()) + }), + + remove: (key) => + Effect.sync(() => { + objects.delete(key) + }), + + exists: (key) => Effect.sync(() => objects.has(key)), + + list: (prefix) => + Effect.sync(() => [...objects.keys()].filter((key) => key.startsWith(prefix))), + } +} + +function objectKey(digest: string): string { + return `objects/${digest}` +} + +function runKey(runId: string): string { + return `runs/${encodeURIComponent(runId)}` +} + +function refPrefix(digest: string): string { + return `refs/${digest}/` +} + +function refKey(digest: string, runId: string): string { + return `${refPrefix(digest)}${encodeURIComponent(runId)}` +} + +function assertDigest(digest: string): Effect.Effect { + if (isSha256Digest(digest)) { + return Effect.void + } + + return Effect.fail(new BlobStoreError(`invalid blob digest ${digest}`)) +} + +export function sha256Hex(body: Uint8Array): Effect.Effect { + const copy = new Uint8Array(body.byteLength) + copy.set(body) + + return Effect.tryPromise({ + try: () => crypto.subtle.digest('SHA-256', copy), + catch: (cause) => new BlobStoreError('SHA-256 failed', cause), + }).pipe(Effect.map((buffer) => hex(new Uint8Array(buffer)))) +} + +function hex(bytes: Uint8Array): string { + let encoded = '' + + for (const byte of bytes) { + encoded += byte.toString(16).padStart(2, '0') + } + + return encoded +} + +function readRunDigests( + bytes: BlobBytes, + runId: string, +): Effect.Effect, BlobStoreError> { + return bytes.read(runKey(runId)).pipe( + Effect.flatMap((body) => { + if (!body) { + return Effect.succeed([]) + } + + return decodeDigestList(new TextDecoder().decode(body)).pipe( + Effect.mapError( + (cause) => new BlobStoreError(`blob references for ${runId} are unreadable`, cause), + ), + ) + }), + ) +} + +function writeRunDigests( + bytes: BlobBytes, + runId: string, + digests: ReadonlyArray, +): Effect.Effect { + return bytes.write(runKey(runId), new TextEncoder().encode(JSON.stringify(digests))) +} diff --git a/packages/core/src/blobs.ts b/packages/core/src/blobs.ts new file mode 100644 index 0000000..1558950 --- /dev/null +++ b/packages/core/src/blobs.ts @@ -0,0 +1,283 @@ +import { Effect, Predicate, Schema } from 'effect' + +import { + BlobStoreError, + isSha256Digest, + sha256Hex, + type BlobStore, + type BlobStoreFailure, +} from './blob-store' +import { utf8JsonBytes } from './limits' +import { isJsonObject, type JsonValue } from './types' + +export const DEFAULT_BLOB_THRESHOLD_BYTES = 64 * 1024 + +const BLOB_KIND = 'blob' +const BLOB_MEDIA_TYPE = 'application/json' +const decodeJsonText = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Json)) + +/** + * Reserved stored shape. A JSON object with exactly these fields is a claim check, + * not caller data. `bytes` is the UTF-8 length of the JSON body named by `sha256`. + */ +export interface BlobRef { + readonly $looms: typeof BLOB_KIND + readonly sha256: string + readonly bytes: number + readonly mediaType: typeof BLOB_MEDIA_TYPE + readonly [key: string]: JsonValue +} + +export function isBlobRef(value: JsonValue): value is BlobRef { + if (!isJsonObject(value) || Object.keys(value).length !== 4) { + return false + } + + const kind = value.$looms + const sha256 = value.sha256 + const bytes = value.bytes + const mediaType = value.mediaType + + return ( + kind === BLOB_KIND && + Predicate.isString(sha256) && + isSha256Digest(sha256) && + Predicate.isNumber(bytes) && + Number.isInteger(bytes) && + bytes >= 0 && + mediaType === BLOB_MEDIA_TYPE + ) +} + +export interface OffloadJsonOptions { + readonly thresholdBytes?: number + /** When false, the root value stays inline even if it is over the threshold. Children are still claimed. */ + readonly claimRoot?: boolean +} + +/** Store `value` and return a reference, even when it is under the offload threshold. */ +export function claimJson( + value: JsonValue, + blobs: BlobStore, +): Effect.Effect { + return putJson(value, blobs) +} + +/** Replace values larger than `thresholdBytes` with {@link BlobRef}s. Bodies are stored by hash. */ +export function offloadJson( + value: JsonValue, + blobs: BlobStore, + options?: OffloadJsonOptions, +): Effect.Effect { + const thresholdBytes = options?.thresholdBytes ?? DEFAULT_BLOB_THRESHOLD_BYTES + + return offloadNode(value, blobs, thresholdBytes, options?.claimRoot ?? true) +} + +/** Replace {@link BlobRef}s with the JSON they name, including references nested inside a body. */ +export function resolveJson( + value: JsonValue, + blobs: BlobStore, +): Effect.Effect { + return resolveNode(value, blobs, new Set()) +} + +/** + * Digests named by `value`, plus digests nested inside those bodies. + * A parent blob that itself contains references still keeps those bodies alive. + */ +export function expandBlobDigests( + blobs: BlobStore, + seeds: ReadonlySet, +): Effect.Effect, BlobStoreFailure> { + if (seeds.size === 0) { + return Effect.succeed(new Set()) + } + + return Effect.gen(function* () { + const all = new Set() + const pending = [...seeds] + + while (pending.length > 0) { + const digest = pending.pop() + + if (digest === undefined || all.has(digest)) { + continue + } + + all.add(digest) + + const body = yield* blobs.get(digest) + const json = yield* decodeBody(body) + const nested = new Set() + + collectBlobDigests(json, nested) + + for (const child of nested) { + if (!all.has(child)) { + pending.push(child) + } + } + } + + return all + }) +} + +/** Record every blob digest embedded in `value`. Does not read blob bodies. */ +export function collectBlobDigests(value: JsonValue, into: Set): void { + if (isBlobRef(value)) { + into.add(value.sha256) + return + } + + if (Array.isArray(value)) { + for (const item of value) { + collectBlobDigests(item, into) + } + + return + } + + if (isJsonObject(value)) { + for (const child of Object.values(value)) { + collectBlobDigests(child, into) + } + } +} + +function offloadNode( + value: JsonValue, + blobs: BlobStore, + thresholdBytes: number, + claim: boolean, +): Effect.Effect { + if (isBlobRef(value)) { + return Effect.succeed(value) + } + + if (Array.isArray(value)) { + return Effect.forEach(value, (item) => offloadNode(item, blobs, thresholdBytes, true)).pipe( + Effect.flatMap((items) => storeIfOver(items, blobs, thresholdBytes, claim)), + ) + } + + if (isJsonObject(value)) { + return offloadObject(value, blobs, thresholdBytes, claim) + } + + return storeIfOver(value, blobs, thresholdBytes, claim) +} + +function offloadObject( + value: { readonly [key: string]: JsonValue }, + blobs: BlobStore, + thresholdBytes: number, + claim: boolean, +): Effect.Effect { + return Effect.gen(function* () { + const next: { [key: string]: JsonValue } = {} + + for (const key of Object.keys(value)) { + const child = value[key] + + if (child === undefined) { + continue + } + + next[key] = yield* offloadNode(child, blobs, thresholdBytes, true) + } + + return yield* storeIfOver(next, blobs, thresholdBytes, claim) + }) +} + +function storeIfOver( + value: JsonValue, + blobs: BlobStore, + thresholdBytes: number, + claim: boolean, +): Effect.Effect { + // Strings only. Objects and arrays stay in place so schema-shaped snapshot state still decodes. + // An event that is still over the byte limit claims its whole payload afterward. + if (!claim || !Predicate.isString(value) || utf8JsonBytes(value) <= thresholdBytes) { + return Effect.succeed(value) + } + + return putJson(value, blobs) +} + +function putJson(value: JsonValue, blobs: BlobStore): Effect.Effect { + const body = new TextEncoder().encode(JSON.stringify(value)) + + return sha256Hex(body).pipe( + Effect.flatMap((sha256) => + blobs.put(sha256, body).pipe( + Effect.map((): BlobRef => ({ + $looms: BLOB_KIND, + sha256, + bytes: body.byteLength, + mediaType: BLOB_MEDIA_TYPE, + })), + ), + ), + ) +} + +function resolveNode( + value: JsonValue, + blobs: BlobStore, + seen: ReadonlySet, +): Effect.Effect { + if (isBlobRef(value)) { + if (seen.has(value.sha256)) { + return Effect.fail(new BlobStoreError(`blob ${value.sha256} refers to itself`)) + } + + const next = new Set(seen) + next.add(value.sha256) + + return blobs.get(value.sha256).pipe( + Effect.flatMap((body) => decodeBody(body)), + Effect.flatMap((decoded) => resolveNode(decoded, blobs, next)), + ) + } + + if (Array.isArray(value)) { + return Effect.forEach(value, (item) => resolveNode(item, blobs, seen)) + } + + if (isJsonObject(value)) { + return resolveObject(value, blobs, seen) + } + + return Effect.succeed(value) +} + +function resolveObject( + value: { readonly [key: string]: JsonValue }, + blobs: BlobStore, + seen: ReadonlySet, +): Effect.Effect { + return Effect.gen(function* () { + const next: { [key: string]: JsonValue } = {} + + for (const key of Object.keys(value)) { + const child = value[key] + + if (child === undefined) { + continue + } + + next[key] = yield* resolveNode(child, blobs, seen) + } + + return next + }) +} + +function decodeBody(body: Uint8Array): Effect.Effect { + return decodeJsonText(new TextDecoder().decode(body)).pipe( + Effect.mapError((cause) => new BlobStoreError('blob body is not JSON', cause)), + ) +} diff --git a/packages/core/src/effects.test.ts b/packages/core/src/effects.test.ts index a545e9d..8f075cf 100644 --- a/packages/core/src/effects.test.ts +++ b/packages/core/src/effects.test.ts @@ -10,6 +10,8 @@ const ctx: EffectContext = { runId: 'run_1', threadId: 'thr_1', causingEventId: 'evt_1', + attributes: {}, + secrets: {}, signal: new AbortController().signal, emit: async () => undefined, } diff --git a/packages/core/src/effects.ts b/packages/core/src/effects.ts index 57035c6..e729599 100644 --- a/packages/core/src/effects.ts +++ b/packages/core/src/effects.ts @@ -2,6 +2,7 @@ import { Data, Effect, Predicate, Schema, type Context } from 'effect' import { EventInputSchema, type EventInput } from './envelope' import { DEFAULT_DEFINITION_VERSION } from './module' +import type { RunAttributes, SecretContext } from './run-context' import { validateInputEffect, type InferDefinedSchema, @@ -187,6 +188,10 @@ export interface EffectContext { readonly threadId: string readonly causingEventId: string readonly signal: AbortSignal + /** Labels safe for every reader of the run. */ + readonly attributes: RunAttributes + /** Bearer material for this run. Never copy it into an emitted event. */ + readonly secrets: SecretContext emit(event: EventInput): Promise } @@ -196,6 +201,8 @@ export interface ScopedEffectContext { readonly threadId: string readonly causingEventId: string readonly signal: AbortSignal + readonly attributes: RunAttributes + readonly secrets: SecretContext emit(event: E): Promise } @@ -215,6 +222,52 @@ export class EffectHandlerError extends Data.TaggedError('EffectHandlerError')<{ } } +export interface EffectFailureOptions { + readonly error: string + /** Fail the attempt even when the retry policy has attempts remaining. */ + readonly nonRetryable?: boolean + /** Wait at least this long before the next attempt. Does not shorten the policy backoff. */ + readonly retryAfterMs?: number +} + +/** A handler's deliberate attempt failure, distinct from an unexpected thrown error. */ +export class EffectFailure extends Data.TaggedError('EffectFailure')<{ + readonly error: string + readonly nonRetryable: boolean + readonly retryAfterMs?: number +}> { + constructor(options: EffectFailureOptions) { + const error = options.error + const nonRetryable = options.nonRetryable === true + + if (options.retryAfterMs === undefined) { + super({ error, nonRetryable }) + } else { + super({ error, nonRetryable, retryAfterMs: options.retryAfterMs }) + } + + this.name = 'EffectFailure' + } +} + +/** Fail the current attempt. The retry policy decides whether another attempt is scheduled. */ +export function failEffect(options: EffectFailureOptions): Effect.Effect { + return Effect.fail(new EffectFailure(options)) +} + +/** Unwrap a handler failure, including one wrapped by `EffectHandlerError`. */ +export function readEffectFailure(cause: unknown): EffectFailure | null { + if (cause instanceof EffectFailure) { + return cause + } + + if (cause instanceof EffectHandlerError) { + return readEffectFailure(cause.cause) + } + + return null +} + export type EffectHandlerResult< R = never, E extends EventInput = EventInput, @@ -266,6 +319,18 @@ function liftHandlerResult( return Effect.succeed(result) } +function invokeHandler( + handler: (input: JsonValue, ctx: EffectContext) => EffectHandlerResult, + input: JsonValue, + ctx: EffectContext, +): Effect.Effect, EffectHandlerError, R> { + try { + return liftHandlerResult(handler(input, ctx)) + } catch (cause) { + return Effect.fail(new EffectHandlerError(cause)) + } +} + /** * Define an effect. With `execute`, a run's actor cell runs the handler at * most once while it is alive; if the cell dies mid-execution, the replacement @@ -327,7 +392,12 @@ export function defineEffect(def: { Effect.gen(function* () { const input = schema ? yield* validateInputEffect(schema, raw) : raw - return handler ? yield* liftHandlerResult(handler(input, ctx)) : noOutput + if (!handler) { + return noOutput + } + + // invokeHandler turns a synchronous throw into EffectHandlerError. + return yield* invokeHandler(handler, input, ctx) }), } } diff --git a/packages/core/src/fold.test.ts b/packages/core/src/fold.test.ts index 74b3654..697d952 100644 --- a/packages/core/src/fold.test.ts +++ b/packages/core/src/fold.test.ts @@ -97,6 +97,7 @@ describe('foldRun', () => { definitionVersion: 'v1', input: null, requestedThreadId: null, + attributes: {}, idempotencyKey: null, rootThreadId: threadId, launchedBy: null, diff --git a/packages/core/src/fold.ts b/packages/core/src/fold.ts index 535df71..3ecc0f6 100644 --- a/packages/core/src/fold.ts +++ b/packages/core/src/fold.ts @@ -5,6 +5,7 @@ import type { EventEnvelope } from './envelope' import { createEffectId } from './ids' import { readRunLaunchSource } from './launcher' import { DEFAULT_DEFINITION_VERSION } from './module' +import { readAttributes } from './run-context' import type { ThreadRecord, OutstandingEffect, RunState, WaitRecord } from './state' import { completedRunStatus, emptyRunState } from './state' import { isTerminalStatus, type ThreadContext, type ThreadDefinition } from './thread' @@ -35,6 +36,28 @@ function readString(obj: { [key: string]: JsonValue }, key: string): string | un return Predicate.isString(value) ? value : undefined } +/** Heartbeat progress already recorded for this effect, if any. */ +function priorHeartbeatDetails(state: RunState, effectId: string): JsonValue | null { + return state.effectExecutions[effectId]?.heartbeatDetails ?? null +} + +/** + * A heartbeat that carries `details` replaces the saved progress. Every other execution event + * keeps it, including the retry that starts the next attempt. + */ +function nextHeartbeatDetails( + state: RunState, + effectId: string, + event: EventEnvelope, + payload: { [key: string]: JsonValue }, +): JsonValue | null { + if (event.type === 'runtime.effect.heartbeat' && Object.hasOwn(payload, 'details')) { + return payload.details ?? null + } + + return priorHeartbeatDetails(state, effectId) +} + function cloneState(state: RunState): RunState { return { runId: state.runId, @@ -50,6 +73,7 @@ function cloneState(state: RunState): RunState { processedIdempotencyKeys: state.processedIdempotencyKeys ? [...state.processedIdempotencyKeys] : [], + attributes: state.attributes ?? {}, } } @@ -127,6 +151,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi rootThreadId, launchedBy: readRunLaunchSource(payload.launchedBy), correlationId: event.correlationId ?? null, + attributes: readAttributes(payload.attributes), } : null @@ -135,6 +160,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi runId: event.runId, rootThreadId: rootThreadId ?? null, startIdentity, + attributes: startIdentity?.attributes ?? {}, status: 'running', } } @@ -290,6 +316,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi deadlineAt: null, startedAt: event.ts, lastHeartbeatAt: null, + heartbeatDetails: priorHeartbeatDetails(state, effectId), lastError: null, }, }, @@ -352,6 +379,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi : sameAttempt ? previous.lastHeartbeatAt : null, + heartbeatDetails: nextHeartbeatDetails(state, effectId, event, payload), lastError: readString(payload, 'error') ?? null, }, }, @@ -384,6 +412,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi deadlineAt: nextAttemptAt, startedAt: null, lastHeartbeatAt: null, + heartbeatDetails: priorHeartbeatDetails(state, effectId), lastError: readString(payload, 'error') ?? 'effect failed', }, }, diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index c1ee11b..4552d7f 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -38,7 +38,11 @@ export * from './serialize' export * from './coalesce' export * from './event-index' export * from './limits' +export * from './blob-store' +export * from './blobs' +export * from './blob-offload' export * from './definition-store' export * from './launcher' +export * from './run-context' export * from './published' export type { DebuggerHost, DebuggerRequest } from './debugger-host' diff --git a/packages/core/src/launcher.ts b/packages/core/src/launcher.ts index 70d7370..7667440 100644 --- a/packages/core/src/launcher.ts +++ b/packages/core/src/launcher.ts @@ -1,5 +1,6 @@ import { Context, Data, Effect, Option, Schema } from 'effect' +import type { RunAttributes, SecretContext } from './run-context' import type { JsonValue, RunStatus } from './types' /** The effect that launched a run, recorded on the launched run's `runtime.run.started`. */ @@ -36,6 +37,8 @@ export interface LaunchRunArgs { * correlation id, or to the launching run's id when that run has none. */ readonly correlationId?: string + readonly attributes?: RunAttributes + readonly secrets?: SecretContext } /** Enough of a run to decide whether a previous fire is still in progress. */ diff --git a/packages/core/src/projection.ts b/packages/core/src/projection.ts index e0fbf05..ed64487 100644 --- a/packages/core/src/projection.ts +++ b/packages/core/src/projection.ts @@ -296,6 +296,7 @@ export function toThreadTree(state: TreeBuildState): ThreadTree { rootThreadId: state.rootThreadId, startIdentity: null, forkedFrom: null, + attributes: {}, threads: state.records, waits: {}, outstandingEffects: [], diff --git a/packages/core/src/protocol.ts b/packages/core/src/protocol.ts index 527ceaf..378c55e 100644 --- a/packages/core/src/protocol.ts +++ b/packages/core/src/protocol.ts @@ -5,6 +5,7 @@ import { WaitConditionSchema, type WaitCondition } from './effects' import { JsonValueSchema } from './envelope' import { RunLaunchSourceSchema } from './launcher' import { DEFAULT_DEFINITION_VERSION } from './module' +import { RunAttributesSchema } from './run-context' const DefinitionVersionSchema = Schema.String.pipe( Schema.withDecodingDefaultKey(Effect.succeed(DEFAULT_DEFINITION_VERSION)), @@ -19,6 +20,7 @@ export const protocolCatalog = defineEventCatalog('runtime', { input: JsonValueSchema, requestedThreadId: Schema.optional(Schema.NullOr(Schema.String)), launchedBy: Schema.optional(RunLaunchSourceSchema), + attributes: Schema.optional(RunAttributesSchema), }), 'run.completed': Schema.Struct({ output: Schema.NullOr(JsonValueSchema), @@ -94,6 +96,8 @@ export const protocolCatalog = defineEventCatalog('runtime', { 'effect.failed': Schema.Struct({ effectId: Schema.String, error: Schema.String, + nonRetryable: Schema.optional(Schema.Boolean), + retryAfterMs: Schema.optional(Schema.Finite), }), 'effect.attempt.started': Schema.Struct({ effectId: Schema.String, @@ -119,6 +123,8 @@ export const protocolCatalog = defineEventCatalog('runtime', { attempt: Schema.Finite, heartbeatAt: Schema.Finite, deadlineAt: Schema.NullOr(Schema.Finite), + /** Progress the next attempt should resume from. Absent on heartbeats that only extend the deadline. */ + details: Schema.optional(JsonValueSchema), }), 'effect.cancel.requested': Schema.Struct({ effectId: Schema.String, diff --git a/packages/core/src/published.ts b/packages/core/src/published.ts index c9836ec..0e2e7f2 100644 --- a/packages/core/src/published.ts +++ b/packages/core/src/published.ts @@ -1,7 +1,8 @@ -import { Schema } from 'effect' +import { Effect, Schema } from 'effect' import { JsonValueSchema } from './envelope' import type { RegisteredDefinition } from './module' +import { RunAttributesSchema } from './run-context' import { jsonSchemaOfInput } from './schema' import { RunStateSchema, type RunState } from './state' @@ -40,6 +41,7 @@ export const RunSummarySchema = Schema.Struct({ status: RunStateSchema.fields.status, kind: Schema.NullOr(Schema.String), definitionName: Schema.NullOr(Schema.String), + attributes: RunAttributesSchema.pipe(Schema.withDecodingDefaultKey(Effect.succeed({}))), }) export type RunSummary = Schema.Schema.Type @@ -50,6 +52,7 @@ export function summarizeRun(state: RunState): RunSummary { status: state.status, kind: state.startIdentity?.kind ?? null, definitionName: state.startIdentity?.definitionName ?? null, + attributes: state.attributes ?? {}, } } diff --git a/packages/core/src/run-context.ts b/packages/core/src/run-context.ts new file mode 100644 index 0000000..6ab8fcd --- /dev/null +++ b/packages/core/src/run-context.ts @@ -0,0 +1,54 @@ +import { Option, Schema, type Effect } from 'effect' + +import type { JsonValue } from './types' + +/** + * Immutable labels for a run, such as tenant, project, or org. + * They are safe for every reader of the run: events, snapshots, summaries, and authorization. + */ +export type RunAttributes = { readonly [key: string]: string } + +/** + * Bearer material for a run, such as an attenuated macaroon. + * It is stored beside the log and is never written to an event, snapshot, or HTTP response. + */ +export type SecretContext = { readonly [key: string]: string } + +export const RunAttributesSchema = Schema.Record(Schema.String, Schema.String) + +export const SecretContextSchema = Schema.Record(Schema.String, Schema.String) + +const decodeAttributes = Schema.decodeUnknownOption(RunAttributesSchema) +const decodeSecretContext = Schema.decodeUnknownOption(SecretContextSchema) + +export interface RunSecretStore { + readonly get: (runId: string) => Effect.Effect + readonly put: (runId: string, secrets: SecretContext) => Effect.Effect +} + +export function hasEntries(record: { readonly [key: string]: string }): boolean { + return Object.keys(record).length > 0 +} + +export function sameStringRecord( + left: { readonly [key: string]: string }, + right: { readonly [key: string]: string }, +): boolean { + const leftKeys = Object.keys(left).toSorted() + const rightKeys = Object.keys(right).toSorted() + + if (leftKeys.length !== rightKeys.length) { + return false + } + + return leftKeys.every((key, index) => key === rightKeys[index] && left[key] === right[key]) +} + +/** Reads a string map from an event payload. Anything else becomes an empty map. */ +export function readAttributes(value: JsonValue | undefined): RunAttributes { + return value === undefined ? {} : Option.getOrElse(decodeAttributes(value), () => ({})) +} + +export function readSecretContext(value: JsonValue): SecretContext { + return Option.getOrElse(decodeSecretContext(value), () => ({})) +} diff --git a/packages/core/src/sqlite-store.ts b/packages/core/src/sqlite-store.ts index 2a6621b..f86b0e7 100644 --- a/packages/core/src/sqlite-store.ts +++ b/packages/core/src/sqlite-store.ts @@ -8,6 +8,7 @@ import { SnapshotPayloadTooLargeError, } from './limits' import type { RunSummary } from './published' +import { RunAttributesSchema, type RunSecretStore } from './run-context' import { SnapshotStoreError, withSnapshotStore, @@ -59,7 +60,13 @@ CREATE TABLE IF NOT EXISTS looms_run_headers ( run_id TEXT PRIMARY KEY, status TEXT NOT NULL, kind TEXT, - definition_name TEXT + definition_name TEXT, + attributes TEXT NOT NULL DEFAULT '{}' +); + +CREATE TABLE IF NOT EXISTS looms_run_secrets ( + run_id TEXT PRIMARY KEY, + body TEXT NOT NULL ); `.trim() @@ -178,6 +185,13 @@ function initSchema(exec: SqliteExec): void { exec.run(sql) } } + + // Databases created before attributes were indexed already have the headers table. + try { + exec.run(`ALTER TABLE looms_run_headers ADD COLUMN attributes TEXT NOT NULL DEFAULT '{}'`) + } catch { + // The column is already present. + } } function parseEvent(body: string): EventEnvelope { @@ -201,9 +215,23 @@ const HeaderRowSchema = Schema.Struct({ status: RunStateSchema.fields.status, kind: Schema.NullOr(Schema.String), definition_name: Schema.NullOr(Schema.String), + attributes: Schema.String, }) const decodeHeaderRow = Schema.decodeUnknownSync(HeaderRowSchema) +const encodeAttributes = Schema.encodeSync(Schema.fromJsonString(RunAttributesSchema)) + +const encodeSecrets = Schema.encodeSync( + Schema.fromJsonString(Schema.Record(Schema.String, Schema.String)), +) + +const decodeAttributesColumn = Schema.decodeUnknownOption( + Schema.fromJsonString(RunAttributesSchema), +) + +const decodeSecretsColumn = Schema.decodeUnknownOption( + Schema.fromJsonString(Schema.Record(Schema.String, Schema.String)), +) function headerFromRow(row: Schema.Schema.Type): RunSummary { return { @@ -211,6 +239,7 @@ function headerFromRow(row: Schema.Schema.Type): RunSumm status: row.status, kind: row.kind, definitionName: row.definition_name, + attributes: Option.getOrElse(decodeAttributesColumn(row.attributes), () => ({})), } } @@ -289,13 +318,20 @@ function createSqliteSnapshotStore(exec: SqliteExec): SnapshotStore { Effect.try({ try: () => { exec.run( - `INSERT INTO looms_run_headers (run_id, status, kind, definition_name) - VALUES (?, ?, ?, ?) + `INSERT INTO looms_run_headers (run_id, status, kind, definition_name, attributes) + VALUES (?, ?, ?, ?, ?) ON CONFLICT (run_id) DO UPDATE SET status = excluded.status, kind = excluded.kind, - definition_name = excluded.definition_name`, - [header.runId, header.status, header.kind, header.definitionName], + definition_name = excluded.definition_name, + attributes = excluded.attributes`, + [ + header.runId, + header.status, + header.kind, + header.definitionName, + encodeAttributes(header.attributes), + ], ) }, catch: (cause) => @@ -305,13 +341,46 @@ function createSqliteSnapshotStore(exec: SqliteExec): SnapshotStore { listHeaders: Effect.try({ try: () => exec - .rows('SELECT run_id, status, kind, definition_name FROM looms_run_headers') + .rows('SELECT run_id, status, kind, definition_name, attributes FROM looms_run_headers') .map((row) => headerFromRow(decodeHeaderRow(row))), catch: (cause) => new SnapshotStoreError('Failed to list run headers', cause), }), } } +const SecretRowSchema = Schema.Struct({ body: Schema.String }) +const decodeSecretRow = Schema.decodeUnknownSync(SecretRowSchema) + +function createSqliteSecrets(exec: SqliteExec): RunSecretStore { + return { + get: (runId) => + Effect.try({ + try: () => { + const [row] = exec.rows('SELECT body FROM looms_run_secrets WHERE run_id = ?', [runId]) + + if (row === undefined) { + return {} + } + + return Option.getOrElse(decodeSecretsColumn(decodeSecretRow(row).body), () => ({})) + }, + catch: (cause) => new EventStoreError(`Failed to read secret context for ${runId}`, cause), + }), + put: (runId, secrets) => + Effect.try({ + try: () => { + exec.run( + `INSERT INTO looms_run_secrets (run_id, body) + VALUES (?, ?) + ON CONFLICT (run_id) DO UPDATE SET body = excluded.body`, + [runId, encodeSecrets(secrets)], + ) + }, + catch: (cause) => new EventStoreError(`Failed to store secret context for ${runId}`, cause), + }), + } +} + function createSqliteEventStore(exec: SqliteExec): EventStore { const waiters = new Map void>>() @@ -580,6 +649,7 @@ function createSqliteEventStore(exec: SqliteExec): EventStore { .map((row) => decodeRunIdRow(row).run_id), catch: (cause) => new EventStoreError('listRuns failed', cause), }), + runSecrets: createSqliteSecrets(exec), } return service diff --git a/packages/core/src/state.ts b/packages/core/src/state.ts index 58d370b..b4da6de 100644 --- a/packages/core/src/state.ts +++ b/packages/core/src/state.ts @@ -9,6 +9,7 @@ import { import { JsonValueSchema } from './envelope' import { RunLaunchSourceSchema, type RunLaunchSource } from './launcher' import { DEFAULT_DEFINITION_VERSION } from './module' +import { RunAttributesSchema, type RunAttributes } from './run-context' import { isTerminalStatus } from './thread' import type { JsonValue, ThreadStatus, RunStatus } from './types' @@ -58,6 +59,8 @@ export interface EffectExecutionRecord { /** When the current attempt started executing, locally or on a worker. */ startedAt: number | null lastHeartbeatAt: number | null + /** Progress reported by the latest heartbeat. Survives retries so the next attempt can resume. */ + heartbeatDetails: JsonValue | null lastError: string | null } @@ -71,6 +74,7 @@ export interface RunStartIdentity { rootThreadId: string launchedBy: RunLaunchSource | null correlationId: string | null + attributes: RunAttributes } /** The run and last sequence a fork copied its history from. */ @@ -85,6 +89,8 @@ export interface RunState { rootThreadId: string | null startIdentity: RunStartIdentity | null forkedFrom: RunForkSource | null + /** Immutable labels copied from `runtime.run.started`. Safe for every reader of the run. */ + attributes: RunAttributes threads: { [threadId: string]: ThreadRecord } waits: { [waitId: string]: WaitRecord } outstandingEffects: readonly OutstandingEffect[] @@ -117,8 +123,10 @@ export const RunStateSchema = Schema.Struct({ correlationId: Schema.NullOr(Schema.String).pipe( Schema.withDecodingDefaultKey(Effect.succeed(null)), ), + attributes: RunAttributesSchema.pipe(Schema.withDecodingDefaultKey(Effect.succeed({}))), }), ).pipe(Schema.withDecodingDefaultKey(Effect.succeed(null))), + attributes: RunAttributesSchema.pipe(Schema.withDecodingDefaultKey(Effect.succeed({}))), forkedFrom: Schema.NullOr(Schema.Struct({ runId: Schema.String, seq: Schema.Finite })).pipe( Schema.withDecodingDefaultKey(Effect.succeed(null)), ), @@ -189,6 +197,9 @@ export const RunStateSchema = Schema.Struct({ lastHeartbeatAt: Schema.NullOr(Schema.Finite).pipe( Schema.withDecodingDefaultKey(Effect.succeed(null)), ), + heartbeatDetails: Schema.NullOr(JsonValueSchema).pipe( + Schema.withDecodingDefaultKey(Effect.succeed(null)), + ), lastError: Schema.NullOr(Schema.String), }), ).pipe(Schema.withDecodingDefaultKey(Effect.succeed({}))), @@ -207,6 +218,7 @@ export function emptyRunState(runId: string): RunState { rootThreadId: null, startIdentity: null, forkedFrom: null, + attributes: {}, threads: {}, waits: {}, outstandingEffects: [], diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 7651bdd..cf46794 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -2,6 +2,7 @@ import { Context, Data, Effect, Layer, Option, Queue, Ref, Stream } from 'effect import { withAssignedSeq, type AppendableEvent, type EventEnvelope } from './envelope' import { assertEventByteLimits, EventPayloadTooLargeError } from './limits' +import { readSecretContext, type RunSecretStore, type SecretContext } from './run-context' export class EventStoreError extends Data.TaggedError('EventStoreError')<{ readonly message: string @@ -288,6 +289,11 @@ export interface EventStore { ) => Stream.Stream readonly listRuns: Effect.Effect + /** + * Run-scoped bearer material. Absent on stores that cannot keep it out of the event log. + * A start that supplies secrets fails closed when this is missing. + */ + readonly runSecrets?: RunSecretStore } /** @@ -321,6 +327,13 @@ interface RunLog { export const makeMemoryEventStore = Effect.gen(function* () { const logs = yield* Ref.make(new Map()) + const secrets = yield* Ref.make(new Map()) + + const runSecrets: RunSecretStore = { + get: (runId) => + Ref.get(secrets).pipe(Effect.map((map) => readSecretContext(map.get(runId) ?? {}))), + put: (runId, value) => Ref.update(secrets, (map) => new Map(map).set(runId, value)), + } const getOrCreate = (map: Map, runId: string): RunLog => { const existing = map.get(runId) @@ -552,6 +565,7 @@ export const makeMemoryEventStore = Effect.gen(function* () { ), listRuns: Ref.get(logs).pipe(Effect.map((map) => [...map.keys()])), + runSecrets, } return service diff --git a/packages/core/src/worker-callback.ts b/packages/core/src/worker-callback.ts index f93722c..cb9a60d 100644 --- a/packages/core/src/worker-callback.ts +++ b/packages/core/src/worker-callback.ts @@ -1,4 +1,6 @@ import type { EventInput } from './envelope' +import type { SecretContext } from './run-context' +import type { JsonValue } from './types' /** Identifies one attempt of an effect executed by an external worker. */ export interface WorkerCallback { @@ -8,8 +10,30 @@ export interface WorkerCallback { /** What an external worker reports about an effect attempt. */ export type WorkerCallbackInput = - | (WorkerCallback & { readonly kind: 'started' | 'heartbeat' | 'cancelled' }) + | (WorkerCallback & { readonly kind: 'started' | 'cancelled' }) + | (WorkerCallback & { readonly kind: 'heartbeat'; readonly details?: JsonValue }) | (WorkerCallback & { readonly kind: 'complete'; readonly events: ReadonlyArray }) - | (WorkerCallback & { readonly kind: 'fail'; readonly error: string }) + | (WorkerCallback & { + readonly kind: 'fail' + readonly error: string + /** When set, the attempt is terminal even if the retry policy has attempts left. */ + readonly nonRetryable?: boolean + /** Extra delay before the next attempt. The policy backoff is never shortened. */ + readonly retryAfterMs?: number + }) export type WorkerCallbackKind = WorkerCallbackInput['kind'] + +/** + * What the host tells the worker about the callback it just sent. + * A queued message cannot be recalled, so `cancelRequested` is how an in-flight attempt learns to stop. + */ +export interface WorkerCallbackResult { + readonly status: 'accepted' | 'stale' | 'cancelled' + readonly cancelRequested: boolean +} + +/** The run's secret context, released only to the attempt that is currently running. */ +export type WorkerSecretsResult = + | { readonly status: 'accepted'; readonly secrets: SecretContext } + | { readonly status: 'stale' | 'cancelled' } diff --git a/packages/react/src/materialize.test.ts b/packages/react/src/materialize.test.ts index 0cc08ab..58e9107 100644 --- a/packages/react/src/materialize.test.ts +++ b/packages/react/src/materialize.test.ts @@ -82,6 +82,7 @@ describe('materializeEvents', () => { kind: 'agent', definitionName: 'assistant', launchedBy, + attributes: {}, }) }) diff --git a/packages/react/src/materialize.ts b/packages/react/src/materialize.ts index 545242c..c92a3f3 100644 --- a/packages/react/src/materialize.ts +++ b/packages/react/src/materialize.ts @@ -2,6 +2,7 @@ import { completedRunStatus, isJsonObject, isJsonString, + readAttributes, readRunLaunchSource, type EventEnvelope, type JsonValue, @@ -29,6 +30,7 @@ function setRunStatus(tables: MaterializedTables, runId: string, status: RunStat kind: existing?.kind ?? null, definitionName: existing?.definitionName ?? null, launchedBy: existing?.launchedBy ?? null, + attributes: existing?.attributes ?? {}, }) } @@ -56,6 +58,7 @@ export function materializeEvents( kind: readString(payload, 'kind') ?? existing?.kind ?? null, definitionName: readString(payload, 'definitionName') ?? existing?.definitionName ?? null, launchedBy: readRunLaunchSource(payload.launchedBy) ?? existing?.launchedBy ?? null, + attributes: readAttributes(payload.attributes), } tables.runs.set(event.runId, next) diff --git a/packages/react/src/react.ts b/packages/react/src/react.ts index 89f2ae1..7b17773 100644 --- a/packages/react/src/react.ts +++ b/packages/react/src/react.ts @@ -13,11 +13,13 @@ import { import { isTerminalStatus, + sameStringRecord, threadTree, toThreadTree, type AnyEventEnvelope, type ProjectionDefinition, type RegisteredEvent, + type RunAttributes, type RunLaunchSource, type RunStatus, type ThreadTree, @@ -232,6 +234,7 @@ export interface RunSummary { readonly startedAt: number | undefined /** The effect in another run that launched this one. */ readonly launchedBy: RunLaunchSource | undefined + readonly attributes: RunAttributes readonly connection: LoomsConnectionStatus } @@ -243,6 +246,7 @@ function shallowEqualSummary(a: RunSummary, b: RunSummary): boolean { a.rootThreadId === b.rootThreadId && a.startedAt === b.startedAt && a.launchedBy === b.launchedBy && + sameStringRecord(a.attributes, b.attributes) && a.connection === b.connection ) } @@ -264,6 +268,7 @@ export function useRunSummary rootThreadId: run?.rootThreadId ?? undefined, startedAt: s.query.startedAt(), launchedBy: run?.launchedBy ?? undefined, + attributes: run?.attributes ?? {}, connection: s.status(), } }, []) diff --git a/packages/react/src/tables.ts b/packages/react/src/tables.ts index b78ff95..9fef0d4 100644 --- a/packages/react/src/tables.ts +++ b/packages/react/src/tables.ts @@ -1,4 +1,4 @@ -import type { RunLaunchSource, RunStatus, ThreadStatus } from '@looms/core' +import type { RunAttributes, RunLaunchSource, RunStatus, ThreadStatus } from '@looms/core' export interface RunRow { runId: string @@ -8,6 +8,7 @@ export interface RunRow { definitionName: string | null /** The effect in another run that launched this one. */ launchedBy: RunLaunchSource | null + attributes: RunAttributes } export interface ThreadRow { diff --git a/packages/runtime/src/blob-offload.test.ts b/packages/runtime/src/blob-offload.test.ts new file mode 100644 index 0000000..a5a9a9e --- /dev/null +++ b/packages/runtime/src/blob-offload.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, test } from 'bun:test' + +import { Effect } from 'effect' + +import { + EventPayloadTooLargeError, + makeMemoryEventStore, + makeMemorySnapshotStore, + memoryBlobStore, +} from '@looms/core' +import { defineWorkflow, workflow } from '@looms/workflow' + +import { createLooms } from './looms' + +const echo = defineWorkflow({ + name: 'blob-echo', + nodes: [{ id: 'done', run: () => null }], +}) + +describe('createLooms blob offload', () => { + test('rejects an event over 1 MiB when no blob store is configured', async () => { + const looms = createLooms({ modules: [workflow({ definitions: [echo] })] }) + const input = { blob: 'x'.repeat(1_500_000) } + + const error = await looms.start(echo, input, { runId: 'too-big' }).then( + () => null, + (cause) => cause, + ) + + expect(error).toBeInstanceOf(EventPayloadTooLargeError) + await looms.stop() + }) + + test('runs a payload over the event limit and reloads it from the offloaded snapshot', async () => { + const inner = await Effect.runPromise(makeMemoryEventStore) + const snapshots = await Effect.runPromise(makeMemorySnapshotStore) + const blobs = memoryBlobStore() + const large = 'x'.repeat(1_500_000) + const input = { blob: large } + + const looms = createLooms({ + modules: [workflow({ definitions: [echo] })], + store: inner, + snapshotStore: snapshots, + blobStore: blobs, + }) + + try { + const started = await looms.start(echo, input, { runId: 'big' }) + expect(started.state.status).toBe('completed') + expect(started.state.startIdentity?.input).toEqual(input) + + const rawEvents = JSON.stringify(await Effect.runPromise(inner.read('big'))) + expect(rawEvents).toContain('"$looms":"blob"') + expect(rawEvents).not.toContain(large) + + const rawSnapshot = await Effect.runPromise(snapshots.loadLatest('big')) + expect(JSON.stringify(rawSnapshot)).toContain('"$looms":"blob"') + expect(JSON.stringify(rawSnapshot)).not.toContain(large) + + const events = await looms.getEvents('big') + expect(JSON.stringify(events)).toContain(large) + } finally { + await looms.stop() + } + + const reloaded = createLooms({ + modules: [workflow({ definitions: [echo] })], + store: inner, + snapshotStore: snapshots, + blobStore: blobs, + rescanTimers: false, + }) + + try { + const state = await reloaded.getRun('big') + expect(state.startIdentity?.input).toEqual(input) + } finally { + await reloaded.stop() + } + }) +}) diff --git a/packages/runtime/src/dispatch-effect.ts b/packages/runtime/src/dispatch-effect.ts index 93a533d..21a8b2e 100644 --- a/packages/runtime/src/dispatch-effect.ts +++ b/packages/runtime/src/dispatch-effect.ts @@ -1,12 +1,14 @@ -import type { Context, ManagedRuntime } from 'effect' -import { Effect } from 'effect' +import type { Context, ManagedRuntime, Types } from 'effect' +import { Effect, Predicate } from 'effect' import { asJson, isPrimitiveEffect, isWaitOnTimer, + readEffectFailure, type ComposedRegistry, type EffectContext, + type EffectFailedPayload, type EventInput, RunLauncherTag, type RegisteredDefinition, @@ -16,6 +18,34 @@ import { import { threadStartedEvents } from './helpers' +function causeMessage(cause: unknown): string { + if (Predicate.isString(cause)) { + return cause + } + + if (cause instanceof Error) { + return cause.message + } + + return String(cause) +} + +function failedAttempt(effectId: string, threadId: string, cause: unknown): EventInput { + const failure = readEffectFailure(cause) + const error = failure?.error ?? causeMessage(cause) + const payload: Types.Mutable = { effectId, error } + + if (failure?.nonRetryable) { + payload.nonRetryable = true + } + + if (failure?.retryAfterMs !== undefined) { + payload.retryAfterMs = failure.retryAfterMs + } + + return { type: 'runtime.effect.failed', payload, threadId } +} + export function dispatchEffect( registry: ComposedRegistry, services: ManagedRuntime.ManagedRuntime, @@ -110,11 +140,7 @@ export function dispatchEffect( if (!handler) { return Effect.succeed([ - { - type: 'runtime.effect.failed', - payload: { effectId: effCtx.effectId, error: `No handler for ${eff.type}` }, - threadId: currentThreadId, - }, + failedAttempt(effCtx.effectId, currentThreadId, `No handler for ${eff.type}`), ]) } @@ -129,18 +155,7 @@ export function dispatchEffect( }) return execution.pipe( - Effect.catch((err) => - Effect.succeed([ - { - type: 'runtime.effect.failed', - payload: { - effectId: effCtx.effectId, - error: err instanceof Error ? err.message : String(err), - }, - threadId: currentThreadId, - }, - ]), - ), + Effect.catch((err) => Effect.succeed([failedAttempt(effCtx.effectId, currentThreadId, err)])), ) } diff --git a/packages/runtime/src/errors.ts b/packages/runtime/src/errors.ts index 8d3611a..2e5b1a9 100644 --- a/packages/runtime/src/errors.ts +++ b/packages/runtime/src/errors.ts @@ -2,6 +2,7 @@ import { Data } from 'effect' import type { JsonValue, + RunAttributes, RunLaunchSource, RunStartIdentity as DurableRunStartIdentity, } from '@looms/core' @@ -15,6 +16,22 @@ export interface RunStartIdentity { readonly idempotencyKey: string | null readonly launchedBy: RunLaunchSource | null readonly correlationId: string | null + readonly attributes: RunAttributes +} + +/** The stored secret context differs. The message never includes the secret values. */ +export class SecretContextConflictError extends Data.TaggedError('SecretContextConflictError')<{ + readonly runId: string + readonly message: string +}> { + constructor(runId: string) { + super({ + runId, + message: `Run "${runId}" already has a different secret context`, + }) + + this.name = 'SecretContextConflictError' + } } export class StartRunConflictError extends Data.TaggedError('StartRunConflictError')<{ diff --git a/packages/runtime/src/filesystem-blob-store.test.ts b/packages/runtime/src/filesystem-blob-store.test.ts new file mode 100644 index 0000000..fbb8cf4 --- /dev/null +++ b/packages/runtime/src/filesystem-blob-store.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, test } from 'bun:test' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { Effect } from 'effect' + +import { offloadJson, resolveJson } from '@looms/core' + +import { filesystemBlobStore } from './filesystem-blob-store' + +describe('filesystemBlobStore', () => { + test('round-trips an offloaded value', async () => { + const directory = await mkdtemp(join(tmpdir(), 'looms-blobs-')) + + try { + const blobs = filesystemBlobStore(directory) + const value = { text: 'z'.repeat(100) } + const stored = await Effect.runPromise(offloadJson(value, blobs, { thresholdBytes: 16 })) + const resolved = await Effect.runPromise(resolveJson(stored, blobs)) + + expect(resolved).toEqual(value) + expect(JSON.stringify(stored)).toContain('"$looms":"blob"') + } finally { + await rm(directory, { recursive: true, force: true }) + } + }) +}) diff --git a/packages/runtime/src/filesystem-blob-store.ts b/packages/runtime/src/filesystem-blob-store.ts new file mode 100644 index 0000000..6fd7119 --- /dev/null +++ b/packages/runtime/src/filesystem-blob-store.ts @@ -0,0 +1,96 @@ +import { mkdir, readdir, readFile, rm, stat, writeFile } from 'node:fs/promises' + +import { Data, Effect, Predicate } from 'effect' + +import { blobStoreFromBytes, BlobStoreError, type BlobBytes, type BlobStore } from '@looms/core' + +class MissingBlobFile extends Data.TaggedError('MissingBlobFile') {} + +/** + * Content-addressed blob store rooted at `directory`. For Bun and Node hosts. + * A Durable Object should use `r2BlobStore` instead. + */ +export function filesystemBlobStore(directory: string): BlobStore { + return blobStoreFromBytes(filesystemBlobBytes(directory)) +} + +function filesystemBlobBytes(directory: string): BlobBytes { + return { + read: (key) => readOptional(`${directory}/${key}`, key), + + write: (key, body) => + Effect.tryPromise({ + try: () => mkdir(`${directory}/${folderOf(key)}`, { recursive: true }), + catch: (cause) => + new BlobStoreError(`Unable to create a directory for blob key ${key}`, cause), + }).pipe( + Effect.andThen( + Effect.tryPromise({ + try: () => writeFile(`${directory}/${key}`, body), + catch: (cause) => new BlobStoreError(`Unable to write blob key ${key}`, cause), + }), + ), + ), + + remove: (key) => + Effect.tryPromise({ + try: () => rm(`${directory}/${key}`, { force: true }), + catch: (cause) => new BlobStoreError(`Unable to delete blob key ${key}`, cause), + }), + + exists: (key) => + Effect.tryPromise({ + try: () => stat(`${directory}/${key}`), + catch: (cause): BlobStoreError | MissingBlobFile => + isMissingFile(cause) + ? new MissingBlobFile() + : new BlobStoreError(`Unable to check blob key ${key}`, cause), + }).pipe( + Effect.as(true), + Effect.catchTag('MissingBlobFile', () => Effect.succeed(false)), + ), + + list: (prefix) => listFolder(directory, prefix), + } +} + +function readOptional(path: string, key: string): Effect.Effect { + return Effect.tryPromise({ + try: () => readFile(path), + catch: (cause): BlobStoreError | MissingBlobFile => + isMissingFile(cause) + ? new MissingBlobFile() + : new BlobStoreError(`Unable to read blob key ${key}`, cause), + }).pipe( + Effect.catchTag('MissingBlobFile', () => Effect.succeed(null)), + Effect.map((body) => (body === null ? null : new Uint8Array(body))), + ) +} + +function listFolder( + directory: string, + prefix: string, +): Effect.Effect, BlobStoreError> { + const folder = prefix.endsWith('/') ? prefix.slice(0, -1) : prefix + + return Effect.tryPromise({ + try: () => readdir(`${directory}/${folder}`), + catch: (cause): BlobStoreError | MissingBlobFile => + isMissingFile(cause) + ? new MissingBlobFile() + : new BlobStoreError(`Unable to list blob keys ${prefix}`, cause), + }).pipe( + Effect.catchTag('MissingBlobFile', () => Effect.succeed([])), + Effect.map((names) => names.map((name) => `${folder}/${name}`)), + ) +} + +function folderOf(key: string): string { + const slash = key.lastIndexOf('/') + + return slash < 0 ? '' : key.slice(0, slash) +} + +function isMissingFile(cause: unknown): boolean { + return Predicate.isObject(cause) && 'code' in cause && cause.code === 'ENOENT' +} diff --git a/packages/runtime/src/helpers.ts b/packages/runtime/src/helpers.ts index c3d4961..91e8c52 100644 --- a/packages/runtime/src/helpers.ts +++ b/packages/runtime/src/helpers.ts @@ -1,4 +1,4 @@ -import { Data, Effect, Layer, Predicate, type Context } from 'effect' +import { Data, Effect, Layer, Predicate, type Context, type Types } from 'effect' import { asJson, @@ -13,6 +13,7 @@ import { validateEventsEffect, validateInputEffect, withSnapshotStore, + hasEntries, type AnyRuntimeModule, type AppendableEvent, type EventCatalog, @@ -23,6 +24,7 @@ import { InvalidEventError, type JsonValue, type OutstandingEffect, + type SecretContext, type RegisteredDefinition, type RunCursor, type RunState, @@ -31,18 +33,36 @@ import { import type { EffectWorkerTask } from './types' +/** What a dispatch carries beyond the effect itself. */ +export interface WorkerTaskExtras { + /** Progress from the previous attempt's last heartbeat. */ + readonly details?: JsonValue | null + readonly secrets?: SecretContext +} + export function workerTask( runId: string, item: OutstandingEffect, attempt: number, + extras: WorkerTaskExtras = {}, ): EffectWorkerTask { - return { + const task: Types.Mutable = { runId, effectId: item.effectId, attempt, type: item.effect.type, input: 'input' in item.effect ? item.effect.input : null, } + + if (extras.details != null) { + task.details = extras.details + } + + if (extras.secrets && hasEntries(extras.secrets)) { + task.secrets = extras.secrets + } + + return task } export class UnknownDefinitionError extends Data.TaggedError('UnknownDefinitionError')<{ diff --git a/packages/runtime/src/index.ts b/packages/runtime/src/index.ts index 1f041f8..6f3d9fa 100644 --- a/packages/runtime/src/index.ts +++ b/packages/runtime/src/index.ts @@ -20,6 +20,7 @@ export { type WakeScheduler, type WorkerCallback, type WorkerCallbackInput, + type WorkerCallbackResult, } from './runtime' export { createLooms, type Looms, type CreateLoomsOptions, type StartResult } from './looms' export { @@ -40,5 +41,6 @@ export { type RunningServer, type LoomsFetchResult, } from './server' +export { filesystemBlobStore } from './filesystem-blob-store' export { createEventStreamResponse, type EventStreamOptions } from './sse' export { notifyObserver, type RuntimeObservation, type RuntimeObserver } from './observer' diff --git a/packages/runtime/src/local-effect-executor.ts b/packages/runtime/src/local-effect-executor.ts index e611713..2e1ba49 100644 --- a/packages/runtime/src/local-effect-executor.ts +++ b/packages/runtime/src/local-effect-executor.ts @@ -11,8 +11,10 @@ import { type EventInput, type OutstandingEffect, type RegisteredDefinition, + type RunAttributes, type RunLauncher, type RunLaunchSource, + type SecretContext, } from '@looms/core' import { dispatchEffect } from './dispatch-effect' @@ -43,6 +45,8 @@ export interface ExecuteLocalEffectsInput { readonly runId: string /** Inherited by runs the effects launch. */ readonly correlationId: string + readonly attributes: RunAttributes + readonly secrets: SecretContext readonly outstanding: ReadonlyArray readonly attempts: ReadonlyMap readonly now: number @@ -56,6 +60,8 @@ function launcherFor( launcher: RunLauncher, runId: string, correlationId: string, + attributes: RunAttributes, + secrets: SecretContext, item: OutstandingEffect, launched: EventEnvelope[], ): RunLauncher { @@ -71,28 +77,37 @@ function launcherFor( launch: (args) => { const targetRunId = args.runId ?? createRunId() - return launcher.launch({ ...args, runId: targetRunId, launchedBy, correlationId }).pipe( - Effect.tap(() => - Effect.sync(() => { - launched.push( - createEvent(runId, { - type: 'runtime.run.launched', - payload: { - runId: targetRunId, - kind: args.kind, - definitionName: args.definitionName, - definitionVersion: args.definitionVersion ?? DEFAULT_DEFINITION_VERSION, - effectId: item.effectId, - }, - threadId: item.threadId, - causationId: item.causingEventId, - correlationId, - origin: { type: 'thread', threadId: item.threadId }, - }), - ) - }), - ), - ) + return launcher + .launch({ + ...args, + runId: targetRunId, + launchedBy, + correlationId, + attributes: args.attributes ?? attributes, + secrets: args.secrets ?? secrets, + }) + .pipe( + Effect.tap(() => + Effect.sync(() => { + launched.push( + createEvent(runId, { + type: 'runtime.run.launched', + payload: { + runId: targetRunId, + kind: args.kind, + definitionName: args.definitionName, + definitionVersion: args.definitionVersion ?? DEFAULT_DEFINITION_VERSION, + effectId: item.effectId, + }, + threadId: item.threadId, + causationId: item.causingEventId, + correlationId, + origin: { type: 'thread', threadId: item.threadId }, + }), + ) + }), + ), + ) }, status: launcher.status, } @@ -183,13 +198,23 @@ export function executeLocalEffects( registry, services, definitions, - launcherFor(launcher, input.runId, input.correlationId, item, launched), + launcherFor( + launcher, + input.runId, + input.correlationId, + input.attributes, + input.secrets, + item, + launched, + ), item.effect, { effectId: item.effectId, runId: input.runId, threadId: item.threadId, causingEventId: item.causingEventId, + attributes: input.attributes, + secrets: input.secrets, signal: controller.signal, emit: appendLive, }, @@ -273,6 +298,7 @@ export function executeLocalEffects( const attempt = input.attempts.get(item.effectId) ?? 1 const payload = Predicate.isObject(outcome.payload) ? outcome.payload : {} const error = Predicate.isString(payload.error) ? payload.error : 'effect failed' + const retryAfterMs = payload.retryAfterMs const failure = planAttemptFailure({ runId: input.runId, @@ -282,6 +308,8 @@ export function executeLocalEffects( now: input.now, retry, origin: { type: 'system' }, + nonRetryable: payload.nonRetryable === true, + retryAfterMs: Predicate.isNumber(retryAfterMs) ? retryAfterMs : undefined, }) if (failure.retrying) { diff --git a/packages/runtime/src/looms.ts b/packages/runtime/src/looms.ts index a1fc06d..613e5ca 100644 --- a/packages/runtime/src/looms.ts +++ b/packages/runtime/src/looms.ts @@ -3,7 +3,9 @@ import { Data, Effect, Predicate } from 'effect' import { EventStoreTag, makeMemoryEventStore, + makeMemorySnapshotStore, snapshotStoreOf, + withBlobOffload, SnapshotStoreTag, type AnyRuntimeModule, type DebuggerHost, @@ -11,6 +13,7 @@ import { type DefinitionRef, type EventEnvelope, type EventInput, + type BlobStore, type EventStore, type EventStoreTrimCoverage, type JsonValue, @@ -27,8 +30,10 @@ import { type EffectWorker, type LoomsRuntime, type RunOperationalStatus, + type StartRunArgs, type WakeScheduler, type WorkerCallbackInput, + type WorkerCallbackResult, } from './runtime' import { createFetchHandler, @@ -51,6 +56,11 @@ export interface CreateLoomsOptions { readonly snapshotEvery?: number /** Defaults to the EventStore's attached store (S2) or an in-memory SnapshotStore. */ readonly snapshotStore?: SnapshotStore + /** + * Content-addressed store for JSON values that would exceed the 1 MiB event limit or the + * 8 MiB snapshot limit. Reads resolve references before folds, projections, and callers see them. + */ + readonly blobStore?: BlobStore readonly trimAfterSnapshot?: { keepSnapshots: number coverage?: (runId: string) => EventStoreTrimCoverage | Promise @@ -95,7 +105,13 @@ export interface Looms { start( definition: TDef, input?: DefinitionInput, - options?: { runId?: string; threadId?: string; idempotencyKey?: string }, + options?: { + runId?: string + threadId?: string + idempotencyKey?: string + attributes?: StartRunArgs['attributes'] + secrets?: StartRunArgs['secrets'] + }, ): Promise /** Start a run by `{ kind, definitionName }` when you only have names (HTTP bodies, CLIs). */ startRun(args: { @@ -106,6 +122,8 @@ export interface Looms { runId?: string threadId?: string idempotencyKey?: string + attributes?: StartRunArgs['attributes'] + secrets?: StartRunArgs['secrets'] }): Promise /** Commit a run's start and schedule a wake without waiting for it. */ launchRun(args: { @@ -116,6 +134,8 @@ export interface Looms { runId?: string threadId?: string idempotencyKey?: string + attributes?: StartRunArgs['attributes'] + secrets?: StartRunArgs['secrets'] }): Promise getRun(runId: string): Promise getEvents(runId: string, options?: { fromSeq?: number; limit?: number }): Promise @@ -135,7 +155,7 @@ export interface Looms { cancel(runId: string, threadId?: string): Promise /** Start a new run that continues where a finished run ended. */ fork(sourceRunId: string, options?: { runId?: string }): Promise - workerCallback(runId: string, input: WorkerCallbackInput): Promise + workerCallback(runId: string, input: WorkerCallbackInput): Promise fetch(req: Request): Promise serve(options?: { port?: number; hostname?: string }): RunningServer stop(): Promise @@ -175,7 +195,7 @@ export function createLooms(options: CreateLoomsOptions = {}): Looms { Effect.gen(function* () { const configuredStore = options.store - const store = !configuredStore + let store = !configuredStore ? yield* makeMemoryEventStore : Predicate.isFunction(configuredStore) ? yield* Effect.tryPromise({ @@ -189,13 +209,22 @@ export function createLooms(options: CreateLoomsOptions = {}): Looms { }) : configuredStore + if (options.blobStore) { + const snapshots = + options.snapshotStore ?? snapshotStoreOf(store) ?? (yield* makeMemorySnapshotStore) + + store = withBlobOffload(store, options.blobStore, { snapshots }) + } + const runtime = createRuntime({ modules, store, maxWakeIterations: options.maxWakeIterations, maxPendingRunOperations: options.maxPendingRunOperations, snapshotEvery: options.snapshotEvery, - snapshotStore: options.snapshotStore ?? snapshotStoreOf(store), + snapshotStore: options.blobStore + ? snapshotStoreOf(store) + : (options.snapshotStore ?? snapshotStoreOf(store)), trimAfterSnapshot: options.trimAfterSnapshot, runCacheSize: options.runCacheSize, scheduler: options.scheduler, @@ -262,6 +291,8 @@ export function createLooms(options: CreateLoomsOptions = {}): Looms { runId: startOpts?.runId, threadId: startOpts?.threadId, idempotencyKey: startOpts?.idempotencyKey, + attributes: startOpts?.attributes, + secrets: startOpts?.secrets, }), ), getRun: (runId) => runEffect((i) => i.runtime.getRun(runId)), diff --git a/packages/runtime/src/planners.test.ts b/packages/runtime/src/planners.test.ts index 53eefd5..30f220d 100644 --- a/packages/runtime/src/planners.test.ts +++ b/packages/runtime/src/planners.test.ts @@ -24,6 +24,7 @@ function runningState(overrides: Partial = {}): RunState { rootThreadId: null, startIdentity: null, forkedFrom: null, + attributes: {}, threads: {}, waits: {}, outstandingEffects: [], @@ -63,6 +64,7 @@ describe('deadline transition planner', () => { nextAttemptAt: null, deadlineAt: 1_001, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, }, }, @@ -95,6 +97,7 @@ describe('deadline transition planner', () => { nextAttemptAt: null, deadlineAt: 900, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, }, future: { @@ -104,6 +107,7 @@ describe('deadline transition planner', () => { nextAttemptAt: null, deadlineAt: 1_200, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, }, }, @@ -143,6 +147,7 @@ describe('deadline transition planner', () => { deadlineAt: 9_000, startedAt: null, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: 'failed', }, }, @@ -186,6 +191,7 @@ describe('worker effect planner', () => { nextAttemptAt: 1_000, deadlineAt: null, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: 'temporary', }, }, @@ -249,6 +255,7 @@ describe('worker effect routing', () => { deadlineAt: null, startedAt: null, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, } : undefined, diff --git a/packages/runtime/src/retry-policy.ts b/packages/runtime/src/retry-policy.ts index eadfe3e..2178dc3 100644 --- a/packages/runtime/src/retry-policy.ts +++ b/packages/runtime/src/retry-policy.ts @@ -18,6 +18,17 @@ export function retryBackoff(retry: RetryPolicy, attempt: number): number { ) } +/** Policy backoff, extended when the failed attempt asked to wait longer. */ +export function retryDelay(retry: RetryPolicy, attempt: number, retryAfterMs?: number): number { + const backoff = retryBackoff(retry, attempt) + + if (retryAfterMs === undefined || !Number.isFinite(retryAfterMs) || retryAfterMs < 0) { + return backoff + } + + return Math.max(backoff, retryAfterMs) +} + export interface AttemptFailurePlan { readonly event: EventEnvelope readonly retrying: boolean @@ -32,10 +43,13 @@ export function planAttemptFailure(input: { readonly now: number readonly retry: RetryPolicy | undefined readonly origin: EventOrigin + /** Customer/domain failures must not be retried even when attempts remain. */ + readonly nonRetryable?: boolean + readonly retryAfterMs?: number }): AttemptFailurePlan { const { runId, item, attempt, error, retry } = input - if (!retry || attempt >= retry.maxAttempts) { + if (input.nonRetryable || !retry || attempt >= retry.maxAttempts) { return { event: createEffectFailedEvent(runId, item, error), retrying: false } } @@ -46,7 +60,7 @@ export function planAttemptFailure(input: { payload: { effectId: item.effectId, attempt, - nextAttemptAt: input.now + retryBackoff(retry, attempt), + nextAttemptAt: input.now + retryDelay(retry, attempt, input.retryAfterMs), error, }, threadId: item.threadId, diff --git a/packages/runtime/src/run-context.test.ts b/packages/runtime/src/run-context.test.ts new file mode 100644 index 0000000..7fcb831 --- /dev/null +++ b/packages/runtime/src/run-context.test.ts @@ -0,0 +1,370 @@ +import { describe, expect, test } from 'bun:test' + +import { Effect } from 'effect' + +import { + RunLauncherTag, + complete, + defineEffect, + defineRuntimeModule, + defineThread, + hasEntries, + invoke, + makeMemoryEventStore, + type EventStore, +} from '@looms/core' + +import { SecretContextConflictError, StartRunConflictError } from './errors' +import { createLooms } from './looms' +import { authDenied, authAllowed, bearerAuth } from './server' + +const SECRET = 'super-secret-token' + +const seen = defineEffect({ + type: 'context.seen', + execute: (_input, ctx) => [ + { + type: 'context.observed', + payload: { + tenant: ctx.attributes.tenant ?? '', + hasSecret: ctx.secrets.token === SECRET, + }, + threadId: ctx.threadId, + }, + ], +}) + +const contextThread = defineThread({ + kind: 'context', + initialState: () => ({ done: false }), + step: (state, event) => (event.type === 'context.observed' ? { ...state, done: true } : state), + effects: (state) => (state.done ? [complete({ ok: true })] : [invoke(seen, {}, 'see')]), +}) + +const context = { kind: 'context', name: 'context', version: 'v1' } + +const launchChild = defineEffect({ + type: 'context.launch', + execute: (_input, ctx) => + Effect.gen(function* () { + const launcher = yield* RunLauncherTag + + yield* launcher.launch({ + runId: `${ctx.runId}-child`, + kind: 'context', + definitionName: 'context', + input: null, + }) + + return [ + { + type: 'context.observed', + payload: { tenant: '', hasSecret: false }, + threadId: ctx.threadId, + }, + ] + }), +}) + +const parentThread = defineThread({ + kind: 'parent', + initialState: () => ({ launched: false }), + step: (state, event) => + event.type === 'context.observed' ? { ...state, launched: true } : state, + effects: (state) => + state.launched ? [complete({ ok: true })] : [invoke(launchChild, {}, 'launch')], +}) + +const contextModule = defineRuntimeModule({ + namespace: 'context', + protocolVersion: '1.0.0', + threads: { context: contextThread, parent: parentThread }, + effects: { seen, launch: launchChild }, + definitions: [ + { kind: 'context', name: 'context', version: 'v1' }, + { kind: 'parent', name: 'parent', version: 'v1' }, + ], +}) + +function containsSecret(encoded: string): boolean { + return encoded.includes(SECRET) +} + +describe('run attributes and secret context', () => { + test('records attributes and keeps secrets out of the log, state, and HTTP responses', async () => { + const dispatched: Array<{ secrets?: { token?: string } }> = [] + + const looms = createLooms({ + modules: [contextModule], + worker: { handles: ['context.seen'], dispatch: (task) => dispatched.push(task) }, + }) + + const started = await looms.start(context, null, { + runId: 'run_context', + attributes: { tenant: 'acme', project: 'billing' }, + secrets: { token: SECRET }, + }) + + expect(started.state.attributes).toEqual({ tenant: 'acme', project: 'billing' }) + expect(containsSecret(JSON.stringify(started))).toBe(false) + + const events = await looms.getEvents('run_context') + const state = await looms.getRun('run_context') + const summaries = await looms.fetch(new Request('http://looms.test/runs/summaries')) + const listed = await summaries?.json() + + expect(containsSecret(JSON.stringify(events))).toBe(false) + expect(containsSecret(JSON.stringify(state))).toBe(false) + expect(containsSecret(JSON.stringify(listed))).toBe(false) + expect(state.attributes).toEqual({ tenant: 'acme', project: 'billing' }) + + expect(events.find((event) => event.type === 'runtime.run.started')?.payload).toMatchObject({ + attributes: { tenant: 'acme', project: 'billing' }, + }) + + const store = await looms.store + expect(store.runSecrets).toBeDefined() + + if (store.runSecrets) { + expect(await Effect.runPromise(store.runSecrets.get('run_context'))).toEqual({ + token: SECRET, + }) + } + + expect(dispatched[0]?.secrets).toEqual({ token: SECRET }) + await looms.stop() + }) + + test('a worker reads the secret context only while its attempt is running', async () => { + const looms = createLooms({ + modules: [contextModule], + worker: { handles: ['context.seen'], dispatch: () => undefined }, + authorize: bearerAuth({ worker: 'worker-token' }), + }) + + await looms.start(context, null, { runId: 'run_worker', secrets: { token: SECRET } }) + const effectId = (await looms.getRun('run_worker')).outstandingEffects[0]?.effectId ?? '' + + const call = async (path: string, init: { token?: string; body?: string } = {}) => { + const response = await looms.fetch( + new Request( + `http://looms.test/runs/run_worker/effects/${encodeURIComponent(effectId)}/${path}`, + { + method: 'POST', + headers: { authorization: `Bearer ${init.token ?? 'worker-token'}` }, + body: init.body ?? '{}', + }, + ), + ) + + return { status: response?.status, headers: response?.headers, body: await response?.json() } + } + + expect((await call('1/secrets')).body).toEqual({ status: 'stale' }) + expect((await call('1/started')).body).toEqual({ status: 'accepted', cancelRequested: false }) + + const accepted = await call('1/secrets') + expect(accepted.body).toEqual({ status: 'accepted', secrets: { token: SECRET } }) + expect(accepted.headers?.get('cache-control')).toBe('no-store') + + expect((await call('1/secrets', { token: 'wrong' })).status).toBe(401) + expect((await call('2/secrets')).body).toEqual({ status: 'stale' }) + + const observed = { + type: 'context.observed', + payload: { tenant: '', hasSecret: true }, + threadId: (await looms.getRun('run_worker')).outstandingEffects[0]?.threadId, + } + + await call('1/complete', { body: JSON.stringify({ events: [observed] }) }) + + expect((await looms.getRun('run_worker')).status).toBe('completed') + expect((await call('1/secrets')).body).toEqual({ status: 'stale' }) + expect(containsSecret(JSON.stringify(await looms.getEvents('run_worker')))).toBe(false) + await looms.stop() + }) + + test('a different attribute or secret context conflicts without revealing the secret', async () => { + const looms = createLooms({ modules: [contextModule] }) + + await looms.start(context, null, { + runId: 'run_once', + idempotencyKey: 'same', + attributes: { tenant: 'acme' }, + secrets: { token: SECRET }, + }) + + const attributeConflict = await looms + .start(context, null, { + runId: 'run_once', + idempotencyKey: 'same', + attributes: { tenant: 'other' }, + secrets: { token: SECRET }, + }) + .then( + () => null, + (error) => error, + ) + + expect(attributeConflict).toBeInstanceOf(StartRunConflictError) + + const secretConflict = await looms + .start(context, null, { + runId: 'run_once', + idempotencyKey: 'same', + attributes: { tenant: 'acme' }, + secrets: { token: 'rotated' }, + }) + .then( + () => null, + (error) => error, + ) + + expect(secretConflict).toBeInstanceOf(SecretContextConflictError) + expect(containsSecret(JSON.stringify(secretConflict))).toBe(false) + expect(JSON.stringify(secretConflict)).not.toContain('rotated') + + const retried = await looms.start(context, null, { + runId: 'run_once', + idempotencyKey: 'same', + attributes: { tenant: 'acme' }, + }) + + expect(retried.runId).toBe('run_once') + + const store = await looms.store + + if (store.runSecrets) { + expect(await Effect.runPromise(store.runSecrets.get('run_once'))).toEqual({ token: SECRET }) + } + + await looms.stop() + }) + + test('authorization sees attributes from the stored run and the start body', async () => { + const decisions: Array<{ attributes?: { tenant?: string } }> = [] + + const looms = createLooms({ + modules: [contextModule], + authorize: (_request, route) => + Effect.sync(() => { + decisions.push(route) + return route.attributes?.tenant === 'acme' ? authAllowed : authDenied(403, 'Forbidden') + }), + }) + + const started = await looms.fetch( + new Request('http://looms.test/runs', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + kind: 'context', + definitionName: 'context', + runId: 'run_auth', + attributes: { tenant: 'acme' }, + }), + }), + ) + + expect(started?.status).toBe(200) + + const read = await looms.fetch(new Request('http://looms.test/runs/run_auth')) + expect(read?.status).toBe(200) + expect(decisions.some((route) => route.attributes?.tenant === 'acme')).toBe(true) + await looms.stop() + }) + + test('authorization reads only the start event to find attributes', async () => { + const memory = await Effect.runPromise(makeMemoryEventStore) + let fullReads = 0 + + const store: EventStore = { + ...memory, + read: (runId, options) => { + if (options?.limit !== 1) { + fullReads += 1 + } + + return memory.read(runId, options) + }, + } + + const readsAtDecision: number[] = [] + + const looms = createLooms({ + modules: [contextModule], + store, + authorize: (_request, route) => + Effect.sync(() => { + if (route.name === 'runs.get') { + readsAtDecision.push(fullReads) + } + + return route.attributes?.tenant === 'acme' ? authAllowed : authDenied(403, 'Forbidden') + }), + }) + + await looms.start(context, null, { runId: 'run_cheap_auth', attributes: { tenant: 'acme' } }) + fullReads = 0 + + const read = await looms.fetch(new Request('http://looms.test/runs/run_cheap_auth')) + + expect(read?.status).toBe(200) + expect(readsAtDecision).toEqual([0]) + await looms.stop() + }) + + test('a launched run and a fork inherit attributes and secrets without logging them', async () => { + const looms = createLooms({ modules: [contextModule] }) + + await looms.start({ kind: 'parent', name: 'parent', version: 'v1' }, null, { + runId: 'run_parent', + attributes: { tenant: 'acme' }, + secrets: { token: SECRET }, + }) + + const deadline = Date.now() + 5_000 + + while ((await looms.getRun('run_parent-child')).rootThreadId === null) { + if (Date.now() > deadline) { + throw new Error('the child run did not start') + } + + await Bun.sleep(5) + } + + const child = await looms.getRun('run_parent-child') + const childEvents = await looms.getEvents('run_parent-child') + expect(child.attributes).toEqual({ tenant: 'acme' }) + expect(containsSecret(JSON.stringify(childEvents))).toBe(false) + + const store = await looms.store + + if (store.runSecrets) { + expect(await Effect.runPromise(store.runSecrets.get('run_parent-child'))).toEqual({ + token: SECRET, + }) + } + + const finished = Date.now() + 5_000 + + while ((await looms.getRun('run_parent')).status !== 'completed') { + if (Date.now() > finished) { + throw new Error('the parent run did not complete') + } + + await Bun.sleep(5) + } + + const forked = await looms.fork('run_parent', { runId: 'run_fork' }) + expect(forked.state.attributes).toEqual({ tenant: 'acme' }) + expect(containsSecret(JSON.stringify(await looms.getEvents('run_fork')))).toBe(false) + + if (store.runSecrets) { + expect(hasEntries(await Effect.runPromise(store.runSecrets.get('run_fork')))).toBe(true) + expect(await Effect.runPromise(store.runSecrets.get('run_fork'))).toEqual({ token: SECRET }) + } + + await looms.stop() + }) +}) diff --git a/packages/runtime/src/run-start.ts b/packages/runtime/src/run-start.ts index 817ac25..5eb16c7 100644 --- a/packages/runtime/src/run-start.ts +++ b/packages/runtime/src/run-start.ts @@ -1,6 +1,6 @@ import { Effect, Predicate } from 'effect' -import type { JsonValue, RunState } from '@looms/core' +import { sameStringRecord, type JsonValue, type RunState } from '@looms/core' import { RuntimeExecutionError, StartRunConflictError, type RunStartIdentity } from './errors' @@ -69,6 +69,10 @@ export function assertSameDurableStart( mismatches.push('correlationId') } + if (!sameStringRecord(existing.attributes ?? {}, requested.attributes)) { + mismatches.push('attributes') + } + return mismatches.length > 0 ? Effect.fail(new StartRunConflictError(runId, requested, existing, mismatches)) : Effect.succeed(existing.rootThreadId) diff --git a/packages/runtime/src/runtime.ts b/packages/runtime/src/runtime.ts index ac0b33b..d48330e 100644 --- a/packages/runtime/src/runtime.ts +++ b/packages/runtime/src/runtime.ts @@ -11,12 +11,16 @@ import { descendantThreadIds, EventStoreTag, foldRun, + hasEntries, isRunTerminal, unavailableRunLauncher, planFork, project, replayTo, + sameStringRecord, summarizeRun, + type EventStore, + type SecretContext, type AnyRuntimeModule, type EventInput, type EventOrigin, @@ -34,6 +38,7 @@ import { RunNotFoundError, RunTerminalError, RuntimeExecutionError, + SecretContextConflictError, StartRunConflictError, } from './errors' import { @@ -79,7 +84,7 @@ export type { StartRunArgs, WakeError, } from './types' -export type { WorkerCallback, WorkerCallbackInput } from '@looms/core' +export type { WorkerCallback, WorkerCallbackInput, WorkerCallbackResult } from '@looms/core' import type { CreateRuntimeOptions, LoomsRuntime, StartRunArgs } from './types' function currentTimeMillis(): number { @@ -111,6 +116,65 @@ const RESCAN_CONCURRENCY = 16 const DEFAULT_KEEP_SNAPSHOTS = 1 const DEFAULT_MAX_PENDING_RUN_OPERATIONS = 256 +interface StartedPayloadInput { + readonly threadId: string + readonly kind: string + readonly definitionName: string + readonly definitionVersion: string + readonly input: StartRunArgs['input'] + readonly requestedThreadId: string | null + readonly launchedBy: StartRunArgs['launchedBy'] + readonly attributes: StartRunArgs['attributes'] +} + +function startedPayload(input: StartedPayloadInput) { + const payload = { + rootThreadId: input.threadId, + kind: input.kind, + definitionName: input.definitionName, + definitionVersion: input.definitionVersion, + input: input.input ?? null, + requestedThreadId: input.requestedThreadId, + } + + const launched = input.launchedBy ? { ...payload, launchedBy: input.launchedBy } : payload + const attributes = input.attributes ?? {} + + return hasEntries(attributes) ? { ...launched, attributes } : launched +} + +function reconcileRunSecrets( + store: EventStore, + runId: string, + requested: SecretContext | undefined, +) { + const next = requested ?? {} + + return Effect.gen(function* () { + const stored = store.runSecrets + + if (!stored) { + if (hasEntries(next)) { + return yield* new RuntimeExecutionError('This event store cannot persist a secret context') + } + + return yield* Effect.void + } + + const current = yield* stored.get(runId) + + if (sameStringRecord(current, next) || requested === undefined) { + return yield* Effect.void + } + + if (!hasEntries(current)) { + return yield* stored.put(runId, next) + } + + return yield* new SecretContextConflictError(runId) + }) +} + export function createRuntime( options: CreateRuntimeOptions, ): LoomsRuntime { @@ -257,24 +321,27 @@ export function createRuntime { status: 'completed', kind: 'workflow', definitionName: 'summarized', + attributes: {}, }, ]) diff --git a/packages/runtime/src/server/auth.ts b/packages/runtime/src/server/auth.ts index 83008f3..4dba778 100644 --- a/packages/runtime/src/server/auth.ts +++ b/packages/runtime/src/server/auth.ts @@ -1,5 +1,7 @@ import { Effect } from 'effect' +import type { RunAttributes } from '@looms/core' + /** * What a route can do. `public` routes (health) are never passed to `authorize`; `read` and * `write` cover run state, `worker` covers external worker callbacks, and `operations` covers @@ -14,6 +16,8 @@ export interface RouteInfo { /** Stable route name, such as `runs.signal` or `effects.retry`. */ readonly name: string readonly runId?: string + /** Present when the run has labels. Safe for the authorizer to branch on. */ + readonly attributes?: RunAttributes } export type AuthDecision = diff --git a/packages/runtime/src/server/fetch-handler.ts b/packages/runtime/src/server/fetch-handler.ts index 831604d..924445a 100644 --- a/packages/runtime/src/server/fetch-handler.ts +++ b/packages/runtime/src/server/fetch-handler.ts @@ -1,6 +1,14 @@ -import { Effect, Predicate } from 'effect' - -import type { EventStore } from '@looms/core' +import { Data, Effect, Option, Predicate, Schema } from 'effect' + +import { + EventStoreTag, + hasEntries, + isJsonObject, + readAttributes, + RunAttributesSchema, + type EventStore, + type RunAttributes, +} from '@looms/core' import type { LoomsRuntime } from '../runtime' import { authResponse, defaultAuthorize, type Authorize } from './auth' @@ -33,11 +41,68 @@ function createRoutes(options: RouteOptions): readonly Route[] { ] } +const StartAttributesSchema = Schema.Struct({ + attributes: Schema.optional(RunAttributesSchema), +}) + +class UnreadableStartBody extends Data.TaggedError('UnreadableStartBody')<{ + readonly cause: unknown +}> {} + +const noAttributes: RunAttributes = {} + +function readRequestAttributes(req: Request): Effect.Effect { + return Effect.tryPromise({ + try: () => req.clone().json(), + catch: (cause) => new UnreadableStartBody({ cause }), + }).pipe( + Effect.map((body) => + Option.match(Schema.decodeUnknownOption(StartAttributesSchema)(body), { + onNone: () => noAttributes, + onSome: (decoded) => decoded.attributes ?? noAttributes, + }), + ), + Effect.catchTag('UnreadableStartBody', () => Effect.succeed(noAttributes)), + ) +} + +/** + * Attributes are immutable and recorded on `runtime.run.started`, which is seq 1 of every + * untrimmed log, forks included. Only a trimmed log needs the snapshot-backed fold. + */ +function readStoredAttributes( + runtime: LoomsRuntime, + store: EventStore, + runId: string, +): Effect.Effect { + const folded = runtime.getRun(runId).pipe( + Effect.provideService(EventStoreTag, store), + Effect.map((state) => state.attributes), + ) + + return store.read(runId, { fromSeq: 1, limit: 1 }).pipe( + Effect.flatMap(([first]) => { + if (first === undefined) { + return Effect.succeed(noAttributes) + } + + if (first.type === 'runtime.run.started' && isJsonObject(first.payload)) { + return Effect.succeed(readAttributes(first.payload.attributes)) + } + + return folded + }), + Effect.catch(() => folded), + ) +} + function authorizeRoute( req: Request, matched: MatchedRoute, authorize: Authorize, -): Effect.Effect { + runtime: LoomsRuntime, + store: EventStore, +): Effect.Effect { const { access, name } = matched.route if (access === 'public') { @@ -46,10 +111,26 @@ function authorizeRoute( const runId = matched.params.runId - return authorize( - req, - Predicate.isString(runId) ? { access, name, runId } : { access, name }, - ).pipe(Effect.map(authResponse)) + return Effect.gen(function* () { + const stored = Predicate.isString(runId) + ? yield* readStoredAttributes(runtime, store, runId) + : noAttributes + + const requested = + matched.route.name === 'runs.start' ? yield* readRequestAttributes(req) : noAttributes + + const attributes = hasEntries(stored) ? stored : requested + + const route = Predicate.isString(runId) + ? hasEntries(attributes) + ? { access, name, runId, attributes } + : { access, name, runId } + : hasEntries(attributes) + ? { access, name, attributes } + : { access, name } + + return yield* authorize(req, route).pipe(Effect.map(authResponse)) + }) } /** Whether the fetch handler answers `pathname`: the HTTP API or the debugger mount. */ @@ -90,7 +171,13 @@ export function createFetchHandler( return new Response('Not Found', { status: 404 }) } - const denied = yield* authorizeRoute(req, matched, authorize) + const denied = yield* authorizeRoute( + req, + matched, + authorize, + options.runtime, + options.store, + ) if (denied) { return denied diff --git a/packages/runtime/src/server/routes/effects.ts b/packages/runtime/src/server/routes/effects.ts index a63c6e5..5d7f2fe 100644 --- a/packages/runtime/src/server/routes/effects.ts +++ b/packages/runtime/src/server/routes/effects.ts @@ -1,6 +1,11 @@ import { Effect, Schema } from 'effect' -import type { WorkerCallbackInput, WorkerCallbackKind } from '@looms/core' +import { + JsonValueSchema, + type JsonValue, + type WorkerCallbackInput, + type WorkerCallbackKind, +} from '@looms/core' import { EventInputBodySchema, provideStore, readJson, type RouteOptions } from '../http' import { integer, oneOf, route, type Route } from '../route-table' @@ -13,9 +18,28 @@ const WORKER_CALLBACK_KINDS = [ 'cancelled', ] as const satisfies readonly WorkerCallbackKind[] +interface FailWorkerCallback { + effectId: string + attempt: number + kind: 'fail' + error: string + nonRetryable?: boolean + retryAfterMs?: number +} + +interface HeartbeatWorkerCallback { + effectId: string + attempt: number + kind: 'heartbeat' + details?: JsonValue +} + const WorkerCallbackBodySchema = Schema.Struct({ events: Schema.optional(Schema.Array(EventInputBodySchema)), error: Schema.optional(Schema.String), + nonRetryable: Schema.optional(Schema.Boolean), + retryAfterMs: Schema.optional(Schema.Finite), + details: Schema.optional(JsonValueSchema), }) function workerCallbackInput( @@ -31,10 +55,47 @@ function workerCallbackInput( switch (params.action) { case 'complete': return { ...callback, kind: 'complete', events: body.events ?? [] } - case 'fail': - return body.error ? { ...callback, kind: 'fail', error: body.error } : null - default: + + case 'fail': { + if (body.error === undefined) { + return null + } + + const failure: FailWorkerCallback = { + ...callback, + kind: 'fail', + error: body.error, + } + + if (body.nonRetryable !== undefined) { + failure.nonRetryable = body.nonRetryable + } + + if (body.retryAfterMs !== undefined) { + failure.retryAfterMs = body.retryAfterMs + } + + return failure + } + + case 'heartbeat': { + const heartbeat: HeartbeatWorkerCallback = { ...callback, kind: 'heartbeat' } + + if (body.details !== undefined) { + heartbeat.details = body.details + } + + return heartbeat + } + + case 'started': + case 'cancelled': return { ...callback, kind: params.action } + + default: { + const exhaustive: never = params.action + return exhaustive + } } } @@ -62,6 +123,25 @@ export function effectRoutes({ runtime, store }: RouteOptions): readonly Route[] ), ), }), + route({ + method: 'POST', + path: '/runs/:runId/effects/:effectId/:attempt/secrets', + name: 'effects.workerSecrets', + access: 'worker', + params: { attempt: integer }, + handle: ({ params }) => + provideStore( + runtime.workerSecrets(params.runId, { + effectId: params.effectId, + attempt: params.attempt, + }), + store, + ).pipe( + Effect.map((result) => + Response.json(result, { headers: { 'cache-control': 'no-store' } }), + ), + ), + }), route({ method: 'POST', path: '/runs/:runId/effects/:effectId/:attempt/:action', @@ -80,8 +160,8 @@ export function effectRoutes({ runtime, store }: RouteOptions): readonly Route[] return Response.json({ error: 'error required' }, { status: 400 }) } - const state = yield* provideStore(runtime.workerCallback(params.runId, input), store) - return Response.json({ runId: params.runId, state }) + const result = yield* provideStore(runtime.workerCallback(params.runId, input), store) + return Response.json(result) }), }), ] diff --git a/packages/runtime/src/server/routes/runs.ts b/packages/runtime/src/server/routes/runs.ts index 1dab2e5..908a7c4 100644 --- a/packages/runtime/src/server/routes/runs.ts +++ b/packages/runtime/src/server/routes/runs.ts @@ -4,6 +4,8 @@ import { createRunId, EventStoreTag, JsonValueSchema, + RunAttributesSchema, + SecretContextSchema, stringifyJson, type EventStore, type JsonValue, @@ -32,6 +34,8 @@ const StartRunBodySchema = Schema.Struct({ input: Schema.optional(JsonValueSchema), runId: Schema.optional(Schema.String), idempotencyKey: Schema.optional(Schema.String), + attributes: Schema.optional(RunAttributesSchema), + secrets: Schema.optional(SecretContextSchema), }) const SignalBodySchema = Schema.Struct({ @@ -185,6 +189,8 @@ export function runRoutes({ runtime, store }: RouteOptions): readonly Route[] { input: body.input ?? null, runId, idempotencyKey, + attributes: body.attributes, + secrets: body.secrets, } if (wantsRunStream(req, url)) { diff --git a/packages/runtime/src/types.ts b/packages/runtime/src/types.ts index 850756b..07e5279 100644 --- a/packages/runtime/src/types.ts +++ b/packages/runtime/src/types.ts @@ -5,7 +5,10 @@ import type { ComposedRegistry, EventEnvelope, EventInput, + WorkerCallback, WorkerCallbackInput, + WorkerCallbackResult, + WorkerSecretsResult, EventStore, EventStoreAppendError, EventStoreError, @@ -14,6 +17,8 @@ import type { EventStoreTrimCoverage, RunLauncher, RunLaunchSource, + RunAttributes, + SecretContext, EventStoreTruncationError, JsonValue, ProjectionDefinition, @@ -92,6 +97,13 @@ export interface EffectWorkerTask { readonly attempt: number readonly type: string readonly input: JsonValue + /** Progress from the previous attempt's last heartbeat, when it reported any. */ + readonly details?: JsonValue + /** + * Present only on the in-memory dispatch. A transport that persists the task must leave this out + * and let the worker read the secret context through an authenticated callback. + */ + readonly secrets?: SecretContext } export interface EffectWorker { @@ -115,6 +127,13 @@ export interface StartRunArgs { launchedBy?: RunLaunchSource /** Recorded on `runtime.run.started` to group related runs. */ correlationId?: string + /** Immutable labels stored on the start event and the run summary. */ + attributes?: RunAttributes + /** + * Bearer material stored beside the log. A retry that sends a different context conflicts. + * Omitting it on a retry keeps the context already stored. + */ + secrets?: SecretContext } export interface StartResult { @@ -203,11 +222,30 @@ export interface LoomsRuntime< sourceRunId: string, options?: { runId?: string }, ): Effect.Effect - /** Applies a worker's report about one effect attempt; stale or foreign reports are ignored. */ + /** + * Applies a worker's report about one effect attempt. A report for another attempt, or one that + * arrives after the attempt settled, is `stale` and changes nothing. + */ workerCallback( runId: string, input: WorkerCallbackInput, - ): Effect.Effect + ): Effect.Effect< + WorkerCallbackResult, + Error | EventStoreAppendError | EventStoreError, + EventStoreTag + > + /** + * The run's secret context for a worker attempt that has started and not settled. Any other + * attempt is `stale`, so a redelivered or superseded message never sees the secrets. + */ + workerSecrets( + runId: string, + callback: WorkerCallback, + ): Effect.Effect< + WorkerSecretsResult, + Error | EventStoreTruncationError | EventStoreTrimmedError | EventStoreError, + EventStoreTag + > readonly listRuns: Effect.Effect /** Indexed run rows. Folds a run only when its header has not been written yet. */ readonly listRunSummaries: Effect.Effect< diff --git a/packages/runtime/src/wake-engine.ts b/packages/runtime/src/wake-engine.ts index 341c9d6..120f3b3 100644 --- a/packages/runtime/src/wake-engine.ts +++ b/packages/runtime/src/wake-engine.ts @@ -139,6 +139,7 @@ export function createWakeEngine(options: WakeEngineOptions): LoomsRuntime['wake return yield* Effect.gen(function* () { const store = yield* EventStoreTag + const secrets = store.runSecrets ? yield* store.runSecrets.get(runId) : {} const runInContext = Effect.runPromiseWith(yield* Effect.context()) executionContext.liveStore = store @@ -247,6 +248,7 @@ export function createWakeEngine(options: WakeEngineOptions): LoomsRuntime['wake registry, worker: options.worker, now, + secrets, }) if (workerPlans.length > 0) { @@ -388,6 +390,8 @@ export function createWakeEngine(options: WakeEngineOptions): LoomsRuntime['wake { runId, correlationId: cursor.state.startIdentity?.correlationId ?? runId, + attributes: cursor.state.attributes ?? {}, + secrets, outstanding, attempts, now, diff --git a/packages/runtime/src/worker-callbacks.ts b/packages/runtime/src/worker-callbacks.ts index b63c9a0..e49e247 100644 --- a/packages/runtime/src/worker-callbacks.ts +++ b/packages/runtime/src/worker-callbacks.ts @@ -9,7 +9,10 @@ import { type OutstandingEffect, type RetryPolicy, type RunState, + type WorkerCallback, type WorkerCallbackInput, + type WorkerSecretsResult, + type WorkerCallbackResult, } from '@looms/core' import { materializeEffectOutcome } from './helpers' @@ -61,9 +64,28 @@ export interface WorkerCallbacksOptions { interface WorkerCallbackPlan { readonly events: ReadonlyArray readonly notices: ReadonlyArray + readonly result: WorkerCallbackResult } -const IGNORED: WorkerCallbackPlan = { events: [], notices: [] } +const STALE: WorkerCallbackPlan = { + events: [], + notices: [], + result: { status: 'stale', cancelRequested: false }, +} + +const CANCELLED: WorkerCallbackPlan = { + events: [], + notices: [], + result: { status: 'cancelled', cancelRequested: true }, +} + +function accepted( + events: ReadonlyArray, + notices: ReadonlyArray = [], + cancelRequested = false, +): WorkerCallbackPlan { + return { events, notices, result: { status: 'accepted', cancelRequested } } +} /** * Turn a worker callback into events against the current durable state. Callbacks for another @@ -88,6 +110,15 @@ function planWorkerCallback(input: { execution?.attempt === callback.attempt && execution.status === 'cancel_requested' + // A message already handed to a worker cannot be pulled back. Tell it to stop instead. + if ( + execution?.attempt === callback.attempt && + execution.status === 'cancel_requested' && + callback.kind !== 'cancelled' + ) { + return Effect.succeed(CANCELLED) + } + if ( !execution || execution.attempt !== callback.attempt || @@ -97,10 +128,9 @@ function planWorkerCallback(input: { execution.status === 'retry_wait' || execution.status === 'timed_out' || execution.status === 'ambiguous' || - (callback.kind === 'cancelled' && !isCancellationAcknowledgement) || - (execution.status === 'cancel_requested' && callback.kind !== 'cancelled') + (callback.kind === 'cancelled' && !isCancellationAcknowledgement) ) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } const retry = item ? registry.effects.get(item.effect.type)?.retry : undefined @@ -117,8 +147,8 @@ function planWorkerCallback(input: { case 'started': { return Effect.succeed( item && execution.status === 'dispatched' - ? { events: [workerStartedEvent(runId, item, callback.attempt, retry, now)], notices: [] } - : IGNORED, + ? accepted([workerStartedEvent(runId, item, callback.attempt, retry, now)]) + : STALE, ) } @@ -129,38 +159,47 @@ function planWorkerCallback(input: { !item || (!implicitStart && execution.status !== 'started' && execution.status !== 'heartbeat') ) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } - return Effect.succeed({ - events: [ - ...(implicitStart ? [workerStartedEvent(runId, item, callback.attempt, retry, now)] : []), - createEvent(runId, { - type: 'runtime.effect.heartbeat', - payload: { - effectId: item.effectId, - attempt: callback.attempt, - heartbeatAt: now, - deadlineAt: workerDeadline(retry, execution.startedAt ?? now, now), - }, - threadId: item.threadId, - origin: WORKER_ORIGIN, - }), - ], - notices: [notice('effect.heartbeat')], - }) + const recorded = { + effectId: item.effectId, + attempt: callback.attempt, + heartbeatAt: now, + deadlineAt: workerDeadline(retry, execution.startedAt ?? now, now), + } + + const payload = + callback.details === undefined ? recorded : { ...recorded, details: callback.details } + + return Effect.succeed( + accepted( + [ + ...(implicitStart + ? [workerStartedEvent(runId, item, callback.attempt, retry, now)] + : []), + createEvent(runId, { + type: 'runtime.effect.heartbeat', + payload, + threadId: item.threadId, + origin: WORKER_ORIGIN, + }), + ], + [notice('effect.heartbeat')], + ), + ) } case 'complete': { if (!item) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } return Effect.forEach(callback.events, (event) => materializeEffectOutcome(registry.catalogs, runId, item, event), ).pipe( - Effect.map((outcomes) => ({ - events: [ + Effect.map((outcomes) => + accepted([ ...outcomes, createEvent(runId, { type: 'runtime.effect.completed', @@ -170,15 +209,14 @@ function planWorkerCallback(input: { causationId: item.causingEventId, origin: WORKER_ORIGIN, }), - ], - notices: [], - })), + ]), + ), ) } case 'fail': { if (!item) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } const failure = planAttemptFailure({ @@ -189,38 +227,45 @@ function planWorkerCallback(input: { now, retry, origin: WORKER_ORIGIN, + nonRetryable: callback.nonRetryable, + retryAfterMs: callback.retryAfterMs, }) - return Effect.succeed({ - events: [failure.event], - notices: failure.retrying - ? [ - { - type: 'effect.retry', - runId, - at: now, - effectId: item.effectId, - attempt: callback.attempt, - detail: callback.error, - }, - ] - : [], - }) + return Effect.succeed( + accepted( + [failure.event], + failure.retrying + ? [ + { + type: 'effect.retry', + runId, + at: now, + effectId: item.effectId, + attempt: callback.attempt, + detail: callback.error, + }, + ] + : [], + ), + ) } case 'cancelled': { - return Effect.succeed({ - events: [ - createEvent(runId, { - type: 'runtime.effect.cancelled', - payload: { effectId: callback.effectId, attempt: callback.attempt }, - threadId: item?.threadId ?? null, - effectId: callback.effectId, - origin: WORKER_ORIGIN, - }), - ], - notices: [notice('effect.cancellation')], - }) + return Effect.succeed( + accepted( + [ + createEvent(runId, { + type: 'runtime.effect.cancelled', + payload: { effectId: callback.effectId, attempt: callback.attempt }, + threadId: item?.threadId ?? null, + effectId: callback.effectId, + origin: WORKER_ORIGIN, + }), + ], + [notice('effect.cancellation')], + true, + ), + ) } default: { @@ -242,19 +287,44 @@ export function createWorkerCallbacks(options: WorkerCallbacksOptions) { const committed = yield* repository.commit(store, runId, (cursor) => planWorkerCallback({ runId, state: cursor.state, callback, registry, now }).pipe( - Effect.map((plan) => ({ events: plan.events, value: plan.notices })), + Effect.map((plan) => ({ + events: plan.events, + value: { notices: plan.notices, result: plan.result }, + })), ), ) if (committed.append) { - for (const notice of committed.value) { + for (const notice of committed.value.notices) { notifyObserver(options.observer, notice) } } - return committed.cursor.state + return committed.value.result }), - ).pipe(Effect.andThen(options.wake(runId))) + ).pipe(Effect.tap((result) => (result.status === 'stale' ? Effect.void : options.wake(runId)))) + + const workerSecrets = (runId: string, callback: WorkerCallback) => + Effect.gen(function* () { + const store = yield* EventStoreTag + const { state } = yield* repository.loadCursor(store, runId) + const execution = state.effectExecutions[callback.effectId] + + if (execution?.attempt !== callback.attempt || isRunTerminal(state)) { + return { status: 'stale' } satisfies WorkerSecretsResult + } + + if (execution.status === 'cancel_requested') { + return { status: 'cancelled' } satisfies WorkerSecretsResult + } + + if (execution.status !== 'started' && execution.status !== 'heartbeat') { + return { status: 'stale' } satisfies WorkerSecretsResult + } + + const secrets = store.runSecrets ? yield* store.runSecrets.get(runId) : {} + return { status: 'accepted', secrets } satisfies WorkerSecretsResult + }) - return { workerCallback: applyWorkerCallback } + return { workerCallback: applyWorkerCallback, workerSecrets } } diff --git a/packages/runtime/src/worker-effect-dispatch.ts b/packages/runtime/src/worker-effect-dispatch.ts index 478a88d..9db4d0e 100644 --- a/packages/runtime/src/worker-effect-dispatch.ts +++ b/packages/runtime/src/worker-effect-dispatch.ts @@ -9,6 +9,7 @@ import { type EventEnvelope, type OutstandingEffect, type RunState, + type SecretContext, } from '@looms/core' import { RuntimeExecutionError } from './errors' @@ -103,6 +104,7 @@ export function planWorkerEffects(input: { readonly registry: ComposedRegistry readonly worker?: EffectWorker readonly now: number + readonly secrets?: SecretContext }): ReadonlyArray { return input.state.outstandingEffects .filter((item) => { @@ -147,7 +149,10 @@ export function planWorkerEffects(input: { causationId: item.causingEventId, origin: { type: 'system' }, }), - task: workerTask(input.runId, item, attempt), + task: workerTask(input.runId, item, attempt, { + details: existing?.heartbeatDetails, + secrets: input.secrets, + }), scheduleToStartDeadlineAt: retry?.scheduleToStartTimeoutMs ? input.now + retry.scheduleToStartTimeoutMs : null, diff --git a/packages/runtime/src/worker-protocol.test.ts b/packages/runtime/src/worker-protocol.test.ts index fe4f539..7bf30c8 100644 --- a/packages/runtime/src/worker-protocol.test.ts +++ b/packages/runtime/src/worker-protocol.test.ts @@ -7,6 +7,7 @@ import { defineEffect, defineRuntimeModule, defineThread, + failEffect, invoke, makeMemoryEventStore, type EventEnvelope, @@ -22,7 +23,10 @@ const workerDefinition = { value: { kind: 'worker-test', name: 'worker-test' }, } -function workerModule(retry: RetryPolicy, execute?: () => ReadonlyArray) { +function workerModule( + retry: RetryPolicy, + execute?: () => ReadonlyArray | Effect.Effect, unknown>, +) { const effect = defineEffect({ type: 'worker.do', retry, @@ -418,7 +422,9 @@ describe('durable external effect workers', () => { expect(unauthorized?.status).toBe(401) expect(unknownAction?.status).toBe(404) expect(startedResponse?.status).toBe(200) + expect(await startedResponse?.json()).toEqual({ status: 'accepted', cancelRequested: false }) expect(completedResponse?.status).toBe(200) + expect(await completedResponse?.json()).toEqual({ status: 'accepted', cancelRequested: false }) expect( (await looms.getEvents(started.runId)).some((event) => event.type === 'worker.done'), @@ -703,4 +709,114 @@ describe('durable external effect workers', () => { expect((await looms.getRun(started.runId)).status).toBe('cancelled') await looms.stop() }) + + test('a non-retryable failure spends no further attempts', async () => { + const looms = createLooms({ + modules: [ + workerModule({ maxAttempts: 3, backoffMs: 1 }, () => + failEffect({ error: 'bad input', nonRetryable: true }), + ), + ], + }) + + const started = await looms.start(workerDefinition, {}) + const types = (await looms.getEvents(started.runId)).map((event) => event.type) + + expect(types).toContain('runtime.effect.failed') + expect(types).not.toContain('runtime.effect.retry.scheduled') + expect((await looms.getRun(started.runId)).status).toBe('failed') + await looms.stop() + }) + + test('retryAfterMs extends the policy backoff', async () => { + const looms = createLooms({ + modules: [ + workerModule({ maxAttempts: 3, backoffMs: 10 }, () => + failEffect({ error: 'later', retryAfterMs: 60_000 }), + ), + ], + }) + + const before = Date.now() + const started = await looms.start(workerDefinition, {}) + + const payload = (await looms.getEvents(started.runId)).find( + (event) => event.type === 'runtime.effect.retry.scheduled', + )?.payload + + const nextAttemptAt = + Predicate.isReadonlyObject(payload) && !Array.isArray(payload) + ? payload.nextAttemptAt + : undefined + + expect(nextAttemptAt).toBeGreaterThanOrEqual(before + 60_000) + + await looms.stop() + }) + + test('heartbeat details resume on the next worker attempt', async () => { + const dispatched: EffectWorkerTask[] = [] + + const looms = createLooms({ + modules: [workerModule({ maxAttempts: 3, backoffMs: 5 })], + worker: { dispatch: (task) => dispatched.push(task) }, + }) + + const started = await looms.start(workerDefinition, {}) + const identity = effectIdentity(await looms.getEvents(started.runId)) + + const heartbeat = await looms.workerCallback(started.runId, { + ...identity, + kind: 'heartbeat', + details: { page: 2 }, + }) + + expect(heartbeat).toEqual({ status: 'accepted', cancelRequested: false }) + + const failed = await looms.workerCallback(started.runId, { + ...identity, + kind: 'fail', + error: 'transient', + }) + + expect(failed).toEqual({ status: 'accepted', cancelRequested: false }) + await Bun.sleep(40) + + expect(dispatched.at(-1)?.attempt).toBe(2) + expect(dispatched.at(-1)?.details).toEqual({ page: 2 }) + + const stale = await looms.workerCallback(started.runId, { + ...identity, + kind: 'heartbeat', + details: { page: 9 }, + }) + + expect(stale).toEqual({ status: 'stale', cancelRequested: false }) + await looms.stop() + }) + + test('a heartbeat during cancellation tells the worker to stop', async () => { + const looms = createLooms({ + modules: [workerModule({ maxAttempts: 1, cancellationTimeoutMs: 60_000 })], + worker: { dispatch: () => undefined, cancel: () => undefined }, + }) + + const started = await looms.start(workerDefinition, {}) + const identity = effectIdentity(await looms.getEvents(started.runId)) + await looms.workerCallback(started.runId, { ...identity, kind: 'started' }) + await looms.cancel(started.runId) + + const heartbeat = await looms.workerCallback(started.runId, { + ...identity, + kind: 'heartbeat', + details: { page: 1 }, + }) + + expect(heartbeat).toEqual({ status: 'cancelled', cancelRequested: true }) + + const events = await looms.getEvents(started.runId) + const heartbeats = events.filter((event) => event.type === 'runtime.effect.heartbeat') + expect(heartbeats).toHaveLength(0) + await looms.stop() + }) })