Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
5 changes: 4 additions & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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/<wallet>.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
Expand Down
19 changes: 17 additions & 2 deletions docs/solana-batch.md
Original file line number Diff line number Diff line change
Expand Up @@ -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` |
Expand Down Expand Up @@ -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 `<lock>.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.

Expand Down
2 changes: 2 additions & 0 deletions src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
126 changes: 119 additions & 7 deletions src/solana-batch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
*
Expand All @@ -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:
Expand Down Expand Up @@ -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.
*
Expand Down Expand Up @@ -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;
}

/**
Expand All @@ -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: {},
Expand Down Expand Up @@ -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. */
Expand Down Expand Up @@ -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 ` +
Expand Down Expand Up @@ -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",
Expand All @@ -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<WalletBatch | WalletRefusal> {
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<BatchAttempt> {
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}.
Expand Down Expand Up @@ -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;
Expand Down
Loading
Loading