diff --git a/CHANGELOG.md b/CHANGELOG.md index 8514989..2b0320e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ ## [Unreleased] +### Added — `batch.onLocked`: wait or fail instead of paying exact (issue #96) + +- Batch pays in one process per wallet: the process that opened the channel + holds its store's lock until it exits, and other processes on the wallet + paid `exact` (1.5–2.5x the metered price), seen only as a + `channel_store_locked` fallback. The README now says so. New + `batch.onLocked: "exact" | "wait" | "throw"` (default `"exact"`, unchanged) + and `batch.lockWaitMs` (default 60000): `"wait"` waits for the lock and then + pays with batch, `"throw"` raises the new `BatchChannelLockedError` + (`"unpaid"`, nothing signed), as does `"wait"` once its wait runs out. + Exported: `BatchChannelLockedError`, `DEFAULT_BATCH_LOCK_WAIT_MS`. + ## [3.24.10] - 2026-10-27 ### Fixed — SolanaLLMClient waits as long as LLMClient (issue #93) diff --git a/CLAUDE.md b/CLAUDE.md index 7e7f382..65fc072 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -193,7 +193,10 @@ src/ atomic, so `payWith()` checks `holdsChannelFile()` (pid + token + the lock file's dev:ino:mtime) before EVERY batch payment and pays exact (`channel_store_locked`) when ownership was lost. Only a `WalletBatch` is memoised in `wallet()`; refusals (except `wallet_config_conflict`) and failures are - re-asked next call. `payWith()` re-checks the close fence right before taking `busy`. Match errors from the shared book by `name`, not `instanceof`. One + re-asked next call. A store another live process owns is `channel_store_locked`: exact by default; +`batch.onLocked: "wait"` re-asks `wallet()` every 500 ms up to `lockWaitMs` (one `backoff` event), `"throw"` (or a +wait run out, or a lost lock when not `"exact"`) → `{kind:"failed"}` with `BatchChannelLockedError` (`"unpaid"`, +nothing signed). The lock is per process, not per call (issue #96 tracks sharing). `payWith()` re-checks the close fence right before taking `busy`. Match errors from the shared book by `name`, not `instanceof`. One batch request in flight per wallet; concurrent calls pay `exact`, they do not queue (they only wait out a 429 cooldown). `parseRetryAfter` drops a non-finite or > 1 day value (default backoff); `coolDown()` caps `wallet.cooldownUntil` at now + `maxWaitMs`, while the call itself weighs the full diff --git a/README.md b/README.md index 29107a0..a72c429 100644 --- a/README.md +++ b/README.md @@ -723,6 +723,7 @@ await client.closeBatchChannel(); - **Never pays twice.** Anything that goes wrong before a batch payment is sent falls back to `exact`. Once one is sent, a call whose payment may have been charged throws `BatchPaymentUnresolvedError` and is never paid again (no new payment, no `exact`, no fallback model). - **Non-streaming chat only.** `chat`, `chatCompletion`, `smartChat` and `smartChatCompletion` use batch; `stream()` and media jobs pay `exact`. - **One channel per wallet.** All clients for a wallet share one channel, saved to `~/.blockrun/solana-batch/.json` (mode `0600`) so a restart reuses it. +- **One process per wallet.** The process that opens the channel holds its store's lock until it exits. Other processes on that wallet (worker pools, PM2/cluster workers, parallel runs) pay `exact`, reported as `channel_store_locked`. Give each worker its own wallet, or route calls through one process. `batch.onLocked: "wait"` waits up to `batch.lockWaitMs` (default 60 s) for the lock, and `"throw"` raises `BatchChannelLockedError` (unpaid, nothing signed) instead of paying `exact`. - **Observable.** Every fallback, backoff, recovery and re-read writes one stderr line; `batch.onEvent` and `client.getBatchStats()` expose the same events and counters. The full runtime contract (the answer-by-answer table, payment-identifier diff --git a/docs/solana-batch.md b/docs/solana-batch.md index 4360ee8..9f94233 100644 --- a/docs/solana-batch.md +++ b/docs/solana-batch.md @@ -312,7 +312,7 @@ stderr, each time it happens: | Event type | `reason` values | | --- | --- | -| `backoff` | `rate_limited`, `cooldown`, `in_doubt` | +| `backoff` | `rate_limited`, `cooldown`, `in_doubt`, `channel_store_locked` (`onLocked: "wait"`) | | `recovered` | the backoff reason it followed (`rate_limited`, `cooldown`, `in_doubt`), `served_unpaid_on_rechallenge`, or `batch_cancelled` | | `fallback` | `not_offered`, `client_signed_not_supported`, `channel_busy`, `untrusted_operator`, `deposit_over_cap`, `channel_pending`, `closed_during_call`, `peer_dependency_missing`, `payment_creation_failed`, `wallet_config_conflict`, `channel_store_locked`, `deposit_journal_unreadable`, `deposit_journal_failed`, `channel_resync_pending`, `channel_resync_failed`, `channel_unreadable`, `payment_required`, `admission_refused`, `rate_limited`, or one of the recognised refusal codes (`batch_payer_not_allowed`, `batch_payer_not_admitted`, `batch_admission_paused`, `batch_server_signed_only`, `batch_unavailable`, `batch_channel_limit`, `PAYMENT_VERIFICATION_UNAVAILABLE`) | | `resync` | `deposit_unanswered`, `deposit_failed`, `deposit_rate_limited`, `deposit_refused`, `receipt_missing`, `receipt_unreconciled`, `orphaned_deposit`, `deposit_unrecorded`, `channel_unusable` | @@ -356,7 +356,22 @@ holding its pid, in the same format earlier releases write and read, so a process still on an older version that shares the channel store (say, during a rolling upgrade) sees the lock as live too; a random ownership token sits beside it in `.owner`. Other processes pay `exact` until that process -exits. A lock that names the current process but was not taken by this SDK +exits (`channel_store_locked`), so a worker pool, PM2/cluster workers or +parallel runs on one wallet get batch in one process only: give each worker +its own wallet, or route calls through one process. + +`batch.onLocked` picks what such a call does instead: + +| `onLocked` | A call while another process owns the store | +| --- | --- | +| `"exact"` (default) | pays `exact`; a `fallback` event, reason `channel_store_locked` | +| `"wait"` | one `backoff` event (`channel_store_locked`, `retryAfterMs` = `lockWaitMs`), then tries the lock every 500 ms; pays with batch once the owner exits, or raises `BatchChannelLockedError` when `batch.lockWaitMs` (default 60000) runs out | +| `"throw"` | raises `BatchChannelLockedError` at once | + +`BatchChannelLockedError` (`wallet`, `ownerPid`, `waitedMs`) is raised before +anything is signed: its retry disposition is `"unpaid"`, so `fallbackModels` +and `smartChat()` may move on (and meet the same lock). A process that loses +its lock to another owner raises it too unless `onLocked` is `"exact"`. A lock that names the current process but was not taken by this SDK version is never treated as stale; if no process uses it (say, a crashed process whose pid was reused), delete the `.lock` and `.lock.owner` files. diff --git a/src/index.ts b/src/index.ts index 97e5066..c8a7f90 100644 --- a/src/index.ts +++ b/src/index.ts @@ -220,8 +220,10 @@ export { BLOCKRUN_SOL_OPERATOR, DEFAULT_BATCH_RATE_LIMIT, DEFAULT_BATCH_IN_DOUBT, + DEFAULT_BATCH_LOCK_WAIT_MS, BatchPaymentUnresolvedError, BatchCloseDeferredError, + BatchChannelLockedError, type BatchUnresolvedReason, type BatchPayloadKind, type SolanaBatchOptions, diff --git a/src/solana-batch.ts b/src/solana-batch.ts index eb99bec..1156a14 100644 --- a/src/solana-batch.ts +++ b/src/solana-batch.ts @@ -114,6 +114,27 @@ export interface SolanaBatchOptions { * by scanning the chain for it. */ channelStore?: string | false; + /** + * What a call does when another live process owns this wallet's channel + * store (`channel_store_locked`). One process per wallet pays with batch: + * it holds the store's lock until it exits, so worker pools, PM2/cluster + * workers or parallel runs on one wallet must each use their own wallet + * (or route their calls through one process). + * - `"exact"` (default): the call pays with `exact`, as before. + * - `"wait"`: the call waits up to {@link lockWaitMs} for the lock (the + * owner has to exit), then pays with batch, or raises + * {@link BatchChannelLockedError} once the wait runs out. + * - `"throw"`: the call raises {@link BatchChannelLockedError} at once. + * + * The error is raised before anything is signed, so it is `"unpaid"`: + * `fallbackModels` (and `smartChat()`'s chain) may move on, and will meet + * the same lock. A process that loses its lock to another owner raises it + * too unless `onLocked` is `"exact"` (waiting cannot give it back). + * Locks between clients of one process do not apply: they share a channel. + */ + onLocked?: "exact" | "wait" | "throw"; + /** Most time, in ms, one call waits for the lock under `onLocked: "wait"`. Default 60000. */ + lockWaitMs?: number; /** * How a call rides out an HTTP 429 (rate limited, or channel capacity * exhausted) on a batch payment before it pays with `exact`. The call waits @@ -171,6 +192,12 @@ export const DEFAULT_BATCH_RATE_LIMIT = { maxAttempts: 3, maxWaitMs: 60_000 } as /** Default {@link SolanaBatchOptions.inDoubt}. */ export const DEFAULT_BATCH_IN_DOUBT = { maxWaitMs: 300_000 } as const; +/** Default {@link SolanaBatchOptions.lockWaitMs}. */ +export const DEFAULT_BATCH_LOCK_WAIT_MS = 60_000; + +/** How often a call under `onLocked: "wait"` tries the lock again. */ +const LOCK_POLL_MS = 500; + /** * One observable batch-settlement event. * @@ -185,7 +212,10 @@ export const DEFAULT_BATCH_IN_DOUBT = { maxWaitMs: 300_000 } as const; * (`reason: "rate_limited"`) or because another call for the wallet hit one * (`reason: "cooldown"`), or before each replay of a payment in doubt that * carries a payment identifier (`reason: "in_doubt"`, or `rate_limited` - * after a 429), with the answer's `status` and the wait in `retryAfterMs`. + * after a 429), with the answer's `status` and the wait in `retryAfterMs`; + * or, once, when a call starts waiting up to `lockWaitMs` for another + * process's channel store lock (`reason: "channel_store_locked"`, + * `onLocked: "wait"`). * - `recovered`: a call that backed off was then paid with batch (`reason: * "in_doubt"` when the replay of a payment in doubt resolved it), or its * fresh challenge after the wait was served without a payment (`reason: @@ -385,6 +415,30 @@ export class BatchCloseDeferredError extends BlockrunError { } } +/** + * Batch is on, another owner holds this wallet's channel store, and the + * client's `batch.onLocked` is `"wait"` (the wait ran out) or `"throw"`. + * Raised before anything is signed: nothing was paid for the call, and its + * {@link RetryDisposition} is `"unpaid"`. + */ +export class BatchChannelLockedError extends BlockrunError { + /** The paying wallet's address. */ + readonly wallet: string; + /** The live owner's pid, when the SDK knows it. */ + readonly ownerPid?: number; + /** How long this call waited for the lock, in ms (0 unless `onLocked` is `"wait"`). */ + readonly waitedMs: number; + + constructor(init: { wallet: string; detail: string; ownerPid?: number; waitedMs: number }) { + super(`batch-settlement unavailable for wallet ${init.wallet}: ${init.detail}`); + this.name = "BatchChannelLockedError"; + this.wallet = init.wallet; + if (init.ownerPid !== undefined) this.ownerPid = init.ownerPid; + this.waitedMs = init.waitedMs; + withDisposition(this, "unpaid"); + } +} + /** * The outcome of one batch attempt. * @@ -2541,6 +2595,8 @@ function intentTarget(intent: DepositIntent): ResyncTarget { interface WalletRefusal { reason: "wallet_config_conflict" | "channel_store_locked" | "deposit_journal_unreadable"; detail: string; + /** `channel_store_locked`: the live owner's pid. */ + ownerPid?: number; } /** @@ -2558,6 +2614,8 @@ export class SolanaBatchPayer { private readonly maxAttempts: number; private readonly maxWaitMs: number; private readonly inDoubtMaxWaitMs: number; + private readonly onLocked: "exact" | "wait" | "throw"; + private readonly lockWaitMs: number; private readonly counters: SolanaBatchStats = { fallbacks: 0, fallbacksByReason: {}, @@ -2593,6 +2651,14 @@ export class SolanaBatchPayer { if (!Number.isFinite(this.inDoubtMaxWaitMs) || this.inDoubtMaxWaitMs < 0) { throw new Error("batch.inDoubt.maxWaitMs must be a finite number of ms >= 0"); } + this.onLocked = init.options.onLocked ?? "exact"; + if (this.onLocked !== "exact" && this.onLocked !== "wait" && this.onLocked !== "throw") { + throw new Error('batch.onLocked must be "exact", "wait" or "throw"'); + } + this.lockWaitMs = init.options.lockWaitMs ?? DEFAULT_BATCH_LOCK_WAIT_MS; + if (!Number.isFinite(this.lockWaitMs) || this.lockWaitMs < 0) { + throw new Error("batch.lockWaitMs must be a finite number of ms >= 0"); + } } /** A copy of this client's batch counters. */ @@ -2636,6 +2702,7 @@ export class SolanaBatchPayer { if (owner !== undefined) { return { reason: "channel_store_locked", + ownerPid: owner, detail: owner === process.pid ? `channel store ${file} is in use by another copy of @blockrun/llm in this process ` + @@ -3064,13 +3131,24 @@ export class SolanaBatchPayer { const unavailable = batchUnavailable(paymentRequired); if (unavailable) return this.fallback(unavailable); let wallet: WalletBatch | WalletRefusal; + let waitedMs = 0; try { wallet = await this.wallet(); + if ("reason" in wallet && wallet.reason === "channel_store_locked" && this.onLocked === "wait") { + const started = Date.now(); + wallet = await this.waitForLock(wallet); + waitedMs = Date.now() - started; + } } catch (err) { // Nothing was sent (say, the channel store's directory is not writable). return this.fallback(creationFailure(err)); } - if ("reason" in wallet) return this.fallback({ reason: wallet.reason, detail: wallet.detail }); + if ("reason" in wallet) { + if (wallet.reason === "channel_store_locked" && this.onLocked !== "exact") { + return this.lockedOut(wallet.detail, wallet.ownerPid, waitedMs); + } + return this.fallback({ reason: wallet.reason, detail: wallet.detail }); + } if (closesAtStart !== undefined && wallet.closes !== closesAtStart) { return this.fallback({ reason: "closed_during_call", @@ -3087,6 +3165,42 @@ export class SolanaBatchPayer { } } + /** + * `onLocked: "wait"`: try the wallet's channel store again every + * {@link LOCK_POLL_MS} until its owner lets go of it or `lockWaitMs` runs + * out. Reported once as a `backoff` event (`channel_store_locked`). + */ + private async waitForLock(refusal: WalletRefusal): Promise { + const deadline = Date.now() + this.lockWaitMs; + let current: WalletBatch | WalletRefusal = refusal; + if (this.lockWaitMs > 0) { + await this.report({ + type: "backoff", + reason: "channel_store_locked", + retryAfterMs: this.lockWaitMs, + detail: refusal.detail, + }); + } + while ("reason" in current && current.reason === "channel_store_locked") { + const left = deadline - Date.now(); + if (left <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, left)); + current = await this.wallet(); + } + return current; + } + + /** `onLocked` other than `"exact"`: the call's `"unpaid"` error, nothing signed. */ + private async lockedOut(detail: string, ownerPid: number | undefined, waitedMs: number): Promise { + const error = new BatchChannelLockedError({ + wallet: await this.init.address(), + detail, + waitedMs, + ...(ownerPid !== undefined ? { ownerPid } : {}), + }); + return { kind: "failed", error }; + } + /** * The close fence for a call about to send its first request: how many * closes this wallet has completed so far, to hand to {@link pay}. @@ -3200,11 +3314,9 @@ export class SolanaBatchPayer { } if (wallet.busy) return fallback({ reason: "channel_busy", attempt: budget.attempt }); if (wallet.file && !holdsChannelFile(wallet.file)) { - return fallback({ - reason: "channel_store_locked", - attempt: budget.attempt, - detail: `this process no longer holds the lock on ${wallet.file}; another owner may be paying into its channels`, - }); + const detail = `this process no longer holds the lock on ${wallet.file}; another owner may be paying into its channels`; + if (this.onLocked !== "exact") return this.lockedOut(detail, undefined, 0); + return fallback({ reason: "channel_store_locked", attempt: budget.attempt, detail }); } wallet.busy = true; holding = true; diff --git a/test/unit/solana-batch.test.ts b/test/unit/solana-batch.test.ts index 1c0cdce..61f3a4c 100644 --- a/test/unit/solana-batch.test.ts +++ b/test/unit/solana-batch.test.ts @@ -16,7 +16,9 @@ import { SolanaLLMClient } from "../../src/solana-client"; import { __resetBatchWalletsForTests, __setBatchSleepForTests, + BatchChannelLockedError, BatchCloseDeferredError, + DEFAULT_BATCH_LOCK_WAIT_MS, BatchPaymentUnresolvedError, FileIntentJournal, ChannelResyncRequiredError, @@ -809,6 +811,106 @@ describe("SolanaLLMClient batch-settlement", () => { expect(exact).toHaveBeenCalledTimes(1); }); + describe("onLocked", () => { + const store = () => path.join(tmp, "channels.json"); + let events: SolanaBatchEvent[]; + + beforeEach(() => { + events = []; + vi.spyOn(console, "error").mockImplementation(() => {}); + // Another live process owns the channel file. + fs.writeFileSync(`${store()}.lock`, String(process.ppid)); + }); + + it('"throw" raises an unpaid BatchChannelLockedError before anything is signed', async () => { + const c = client({ onLocked: "throw", onEvent: (e) => events.push(e) }); + const exact = stubExact(c); + gateway.push(() => quote402([exactAccept(), batchAccept(operator.address)])); + + const err = await c.chat("openai/gpt-4o-mini", "gm").catch((e: unknown) => e); + + expect(err).toBeInstanceOf(BatchChannelLockedError); + expect(err).toMatchObject({ wallet: await c.getWalletAddress(), ownerPid: process.ppid, waitedMs: 0 }); + expect((err as Error).message).toContain(`in use by process ${process.ppid}`); + expect(retryDisposition(err)).toBe("unpaid"); + expect(exact).not.toHaveBeenCalled(); + // Only the unpaid challenge was sent. + expect(gatewayCalls).toHaveLength(1); + expect(c.getBatchStats().fallbacks).toBe(0); + }); + + it('"wait" pays with batch once the owner lets go of the lock', async () => { + let polls = 0; + __setBatchSleepForTests(async () => { + // The owner exits after the second poll. + if (++polls === 2) fs.unlinkSync(`${store()}.lock`); + }); + const c = client({ onLocked: "wait", onEvent: (e) => events.push(e) }); + const exact = stubExact(c); + gateway.push( + () => quote402([exactAccept(), batchAccept(operator.address)]), + async (_url, init) => { + const payment = decodePayment(init); + expect(payment.payload.type).toBe("deposit"); + return servedWithVoucher(operator, channelIdOf(payment), 1000n, 1000n); + } + ); + + await expect(c.chat("openai/gpt-4o-mini", "gm")).resolves.toBe("gm"); + + expect(exact).not.toHaveBeenCalled(); + expect(polls).toBe(2); + expect(events).toEqual([ + expect.objectContaining({ type: "backoff", reason: "channel_store_locked", retryAfterMs: DEFAULT_BATCH_LOCK_WAIT_MS }), + ]); + expect(fs.readFileSync(`${store()}.lock`, "utf8")).toBe(String(process.pid)); + }); + + it('"wait" raises BatchChannelLockedError when lockWaitMs runs out, and never pays exact', async () => { + let slept = 0; + __setBatchSleepForTests(async (ms) => { + slept += ms; + vi.setSystemTime(Date.now() + ms); + }); + vi.useFakeTimers({ toFake: ["Date"] }); + try { + const c = client({ onLocked: "wait", lockWaitMs: 2_000, onEvent: (e) => events.push(e) }); + const exact = stubExact(c); + gateway.push(() => quote402([exactAccept(), batchAccept(operator.address)])); + + const err = await c.chat("openai/gpt-4o-mini", "gm").catch((e: unknown) => e); + + expect(err).toBeInstanceOf(BatchChannelLockedError); + expect((err as BatchChannelLockedError).waitedMs).toBeGreaterThanOrEqual(2_000); + expect(slept).toBe(2_000); + expect(exact).not.toHaveBeenCalled(); + expect(gatewayCalls).toHaveLength(1); + } finally { + vi.useRealTimers(); + } + }); + + it('"exact" (the default) keeps paying exact and reports the fallback', async () => { + const c = client({ onEvent: (e) => events.push(e) }); + const exact = stubExact(c); + gateway.push( + () => quote402([exactAccept(), batchAccept(operator.address)]), + () => new Response(JSON.stringify(CHAT_OK), { status: 200 }) + ); + + await c.chat("openai/gpt-4o-mini", "gm"); + + expect(exact).toHaveBeenCalledTimes(1); + expect(events).toEqual([expect.objectContaining({ type: "fallback", reason: "channel_store_locked" })]); + }); + + it("rejects an unknown onLocked or a bad lockWaitMs", () => { + expect(() => client({ onLocked: "queue" as never })).toThrow(/batch.onLocked/); + expect(() => client({ lockWaitMs: -1 })).toThrow(/batch.lockWaitMs/); + expect(() => client({ lockWaitMs: Number.POSITIVE_INFINITY })).toThrow(/batch.lockWaitMs/); + }); + }); + describe("channel retention", () => { let events: SolanaBatchEvent[];