From a4cb52c07fb4ca674b65b27b3363e91a7841fd6b Mon Sep 17 00:00:00 2001 From: Russell Dempsey <1173416+SgtPooki@users.noreply.github.com> Date: Fri, 14 Aug 2026 16:58:30 -0400 Subject: [PATCH 1/3] feat: add migrate batched upload with resume Fourth slice of the migrate command split (#652). Drives packed pieces through store, SP-to-SP pull to secondaries, and batched addPieces commits (up to the SDK batch cap), persisting every transition to migrate.db. Commits whose outcome is unknown after a crash are reconciled on resume from the PiecesAdded event, or via findPieceIdsByCid when no tx hash was captured, so a blind on-chain re-add is never issued. pdp-verifier wraps the read-only chain lookups the reconciliation needs (piece id by CID, tx receipt status, PiecesAdded event extraction). --- src/migrate/direct-upload.ts | 616 ++++++++++++++++++ src/migrate/pdp-verifier.ts | 78 +++ .../unit/migrate-direct-upload-flow.test.ts | 467 +++++++++++++ 3 files changed, 1161 insertions(+) create mode 100644 src/migrate/direct-upload.ts create mode 100644 src/migrate/pdp-verifier.ts create mode 100644 src/test/unit/migrate-direct-upload-flow.test.ts diff --git a/src/migrate/direct-upload.ts b/src/migrate/direct-upload.ts new file mode 100644 index 00000000..0987fada --- /dev/null +++ b/src/migrate/direct-upload.ts @@ -0,0 +1,616 @@ +/** + * Direct upload: stream locally packed CARs straight to storage providers and + * batch the on-chain adds. Nothing here requires inbound connectivity. + * + * Per built sub-piece CAR: + * 1. store() the bytes on the primary provider: the piece is now "parked" + * and Curio's GC clock starts. + * 2. Each secondary pulls the piece from the primary's retrieval URL + * (provider-to-provider; the client uploads once). + * 3. Parked pieces accumulate per provider and are flushed through one + * commit() (addPieces) when the batch fills, the GC-window guess nears + * expiry, or the source drains: see gc-window.ts for the scheduling + * rules and why every tie breaks toward flushing early. + * + * Gas is the provider's cost (the provider submits addPieces with the + * client's EIP-712 authorisation), so this loop has no base-fee gate: + * pausing would save the provider gas while running the client's parked + * pieces into GC. + */ + +import { createReadStream } from 'node:fs' +import { unlink } from 'node:fs/promises' +import { Readable } from 'node:stream' +import { findPiece } from '@filoz/synapse-core/sp' +import type { Synapse } from '@filoz/synapse-sdk' +import { CID } from 'multiformats/cid' +import { formatFileSize } from '../utils/cli-helpers.js' +import type { MigrationDB } from './db.js' +import { + collectedCidFromError, + DEFAULT_ASSUMED_WINDOW_MS, + lowerWindowOnGc, + MAX_ADD_PIECES_BATCH, + marginFromConfirmations, + shouldFlush, +} from './gc-window.js' +import { formatDuration, Timer } from './metrics.js' +import { type AddPiecesEvent, dataSetPieceId, fetchAddPiecesEvent, txLanded } from './pdp-verifier.js' +import { log } from './util.js' + +export interface DirectUploadOptions { + /** Initialized Synapse instance (auth, chain, and transport already resolved). */ + synapse: Synapse + /** Number of provider copies (contexts). Default 2: primary + one secondary. */ + copies?: number | undefined + /** Pin specific providers instead of SDK selection. */ + providerIds?: bigint[] | undefined + /** Reuse existing data sets instead of creating new ones. */ + dataSetIds?: bigint[] | undefined + /** Starting GC-window guess; persisted per-provider lowering still applies. */ + assumedWindowMs?: number | undefined + /** Data-set metadata applied when creating or matching contexts. */ + dataSetMetadata?: Record | undefined + /** Route retrieval through the FilBeam egress CDN. */ + withCDN?: boolean | undefined + /** + * Pipeline hook: called when no sub-piece is currently pending. Resolves + * true when new sub-pieces may have appeared (re-poll), false when the + * source is drained. Absent means the source was fully staged up-front. + */ + waitForMore?: (() => Promise) | undefined + /** + * Pipeline hook: when it returns true while the upload queue is empty, + * parked pieces are flushed immediately instead of waiting for a full + * batch or the GC-window timer. The disk-budget gate uses this so blocked + * downloads are never left waiting on a commit the batcher sees no reason + * to hurry. + */ + forceFlush?: (() => boolean) | undefined +} + +/** The storage-context surface the loop drives; narrowed for fakes in tests. */ +export interface UploadContextLike { + providerId: string + serviceURL: string + dataSetId: string | null + store( + data: ReadableStream | Uint8Array, + options: { pieceCid?: unknown; onProgress?: (bytes: number) => void } + ): Promise<{ pieceCid: unknown; size: number }> + /** EIP-712 authorization for pulls/commits of these pieces on this provider. */ + presignForCommit(pieces: Array<{ pieceCid: unknown }>): Promise + pull(options: { + pieces: unknown[] + /** + * Pull source. MUST be the per-piece URL function form: the SDK treats a + * string as a service-URL base and appends its own path, which mangles an + * already-complete piece URL into a source the provider cannot fetch. + */ + from: (pieceCid: unknown) => string + extraData?: unknown + }): Promise<{ + status: 'complete' | 'failed' + pieces: Array<{ pieceCid: unknown; status: 'complete' | 'failed' }> + }> + commit(options: { + pieces: Array<{ pieceCid: unknown; pieceMetadata?: Record }> + onSubmitted?: (txHash: string) => void + }): Promise<{ + txHash: string + pieceIds: bigint[] + dataSetId: bigint + }> + getPieceUrl(pieceCid: unknown): string + /** Probe whether a parked piece is still present (post-GC re-verify). */ + hasPiece(pieceCid: unknown): Promise +} + +export interface DirectUploadDeps { + setup(opts: DirectUploadOptions): Promise<{ contexts: UploadContextLike[] }> + /** Injectable clock so tests can drive the window timer. */ + now(): number + /** Open a built CAR for streaming. Injectable so tests skip the filesystem. */ + openCar(path: string): ReadableStream | Uint8Array + evictCar(path: string): Promise + /** Whether a transaction landed successfully on chain (receipt status 1). */ + txLanded(synapse: Synapse, txHash: string): Promise + /** Canonical on-chain witness for a landed addPieces (see pdp-verifier). */ + fetchAddPiecesEvent(synapse: Synapse, dataSetId: number, txHash: string): Promise + /** On-chain piece id lookup for a data set, or null when absent (see pdp-verifier). */ + dataSetPieceId(synapse: Synapse, dataSetId: number, pieceCid: string): Promise +} + +export const defaultDirectUploadDeps: DirectUploadDeps = { + async setup(opts) { + const contexts = await opts.synapse.storage.createContexts({ + copies: opts.copies ?? 2, + ...(opts.providerIds == null ? {} : { providerIds: opts.providerIds }), + ...(opts.dataSetIds == null ? {} : { dataSetIds: opts.dataSetIds }), + // When targeting existing data sets by ID, metadata is not used for + // matching: pass it only for creation-path selection so an existing + // set with different metadata stays reachable. + ...(opts.dataSetIds == null && opts.dataSetMetadata != null ? { metadata: opts.dataSetMetadata } : {}), + ...(opts.withCDN === true ? { withCDN: true } : {}), + }) + if (contexts.length === 0) throw new Error('no storage contexts resolved') + return { + contexts: contexts.map((ctx): UploadContextLike => { + const serviceURL = ctx.provider.pdp.serviceURL + return { + providerId: String(ctx.provider.id), + serviceURL, + dataSetId: ctx.dataSetId == null ? null : String(ctx.dataSetId), + store: (data, options) => ctx.store(data as never, options as never), + presignForCommit: (pieces) => ctx.presignForCommit(pieces as never), + pull: (options) => ctx.pull(options as never), + commit: (options) => ctx.commit(options as never), + getPieceUrl: (pieceCid) => ctx.getPieceUrl(pieceCid as never), + hasPiece: async (pieceCid) => { + try { + await findPiece({ serviceURL, pieceCid: pieceCid as never, retryCount: 0 }) + return true + } catch { + return false + } + }, + } + }), + } + }, + now: () => Date.now(), + openCar: (path) => Readable.toWeb(createReadStream(path)) as ReadableStream, + evictCar: async (path) => { + try { + await unlink(path) + } catch (err) { + // A resumed run may find the CAR already evicted by a prior run. + if ((err as NodeJS.ErrnoException).code !== 'ENOENT') { + log(`warn: failed to evict cached CAR ${path}: ${(err as Error).message}`) + } + } + }, + txLanded, + fetchAddPiecesEvent, + dataSetPieceId, +} + +export interface DirectUploadSummary { + network: string + providers: Array<{ + providerId: string + role: 'primary' | 'secondary' + dataSetId: string | null + committed: number + collected: number + failed: number + /** + * Pieces whose addPieces outcome is unknown (the attempt was made but + * never confirmed either way). Anything non-zero means the run is + * incomplete: resolve on chain before retrying. + */ + addUnconfirmed: number + flushes: number + assumedWindowMs: number + /** Distinct addPieces transaction hashes behind the committed pieces. */ + txHashes: string[] + }> + storedBytes: number + evictedCars: number +} + +export async function runDirectUpload( + db: MigrationDB, + opts: DirectUploadOptions, + deps: DirectUploadDeps = defaultDirectUploadDeps +): Promise { + const { synapse } = opts + const { contexts } = await deps.setup(opts) + const [primary, ...secondaries] = contexts + if (primary == null) throw new Error('no primary storage context') + + log( + `direct upload to ${contexts.length} provider(s): ` + + contexts.map((c, i) => `${i === 0 ? 'primary' : 'secondary'} ${c.providerId} (${c.serviceURL})`).join(', ') + ) + + const observedCommitMs: number[] = [] + const flushCounts = new Map() + const runTimer = new Timer() + let storedBytes = 0 + + const windowFor = (ctx: UploadContextLike): number => + db.providerWindowMs(ctx.providerId, opts.assumedWindowMs ?? DEFAULT_ASSUMED_WINDOW_MS) + + const flush = async (ctx: UploadContextLike, reason: string): Promise => { + const batch = db.parkedUploads(ctx.providerId).slice(0, MAX_ADD_PIECES_BATCH) + if (batch.length === 0) return + flushCounts.set(ctx.providerId, (flushCounts.get(ctx.providerId) ?? 0) + 1) + const cids = batch.map((b) => b.subPieceCid) + log(`flush [${reason}] provider ${ctx.providerId}: committing ${batch.length} piece(s)`) + // Durable breadcrumb before the attempt: a crash mid-commit must never be + // auto-resolved into a blind re-add. + db.markUploadsAddUnconfirmed(cids, ctx.providerId) + const commitTimer = new Timer() + try { + const result = await ctx.commit({ + pieces: cids.map((cid) => ({ pieceCid: CID.parse(cid) })), + onSubmitted: (txHash) => db.markUploadTxSubmitted(cids, ctx.providerId, txHash), + }) + observedCommitMs.push(commitTimer.stop()) + batch.forEach((b, i) => { + db.markUploadCommitted(b.subPieceCid, ctx.providerId, { + dataSetId: String(result.dataSetId), + pieceId: String(result.pieceIds[i] ?? ''), + txHash: result.txHash, + }) + }) + log(`committed ${batch.length} piece(s) on provider ${ctx.providerId} (data set ${result.dataSetId})`) + } catch (err) { + const message = (err as Error).message ?? String(err) + const gcCid = collectedCidFromError(message) + if (gcCid == null) { + // Not a GC rejection: leave the batch in add_unconfirmed for the + // resume reconciliation: a blind retry could double-add on chain. + log(`error: commit failed on provider ${ctx.providerId} (batch left add_unconfirmed): ${message}`) + return + } + // Curio rejected the batch because a parked piece is gone. The batch is + // atomic and pre-chain, so nothing landed. Lower the window from the + // collected piece's parked age, then re-verify every batch member: + // Curio reports only the FIRST miss. + const collected = batch.find((b) => b.subPieceCid === gcCid) + if (collected == null) { + log(`warn: provider ${ctx.providerId} rejected unknown sub-piece ${gcCid}; re-verifying batch`) + } else { + const age = deps.now() - Date.parse(collected.parkedAt) + const lowered = lowerWindowOnGc(windowFor(ctx), age) + db.lowerProviderWindow(ctx.providerId, lowered) + log( + `GC detected on provider ${ctx.providerId}: ${gcCid} collected after ${formatDuration(age)} parked; ` + + `window lowered to ${formatDuration(lowered)}` + ) + } + for (const b of batch) { + const present = b.subPieceCid !== gcCid && (await ctx.hasPiece(CID.parse(b.subPieceCid))) + if (present) { + db.revertUploadsToParked([b.subPieceCid], ctx.providerId) + } else { + db.markUploadCollected(b.subPieceCid, ctx.providerId) + log(`collected: ${b.subPieceCid} on provider ${ctx.providerId} (will re-store)`) + } + } + } + } + + // Evict staged CARs whose every copy is committed. Runs after every flush, + // not just at run end: the disk high-water mark must track the uncommitted + // window, not the whole migration. The DB keeps car_path after eviction (the + // row is the piece's provenance), so track what this run already unlinked. + const evictedPaths = new Set() + const evictCommitted = async (): Promise => { + for (const path of db.carPathsEvictable()) { + if (evictedPaths.has(path)) continue + await deps.evictCar(path) + evictedPaths.add(path) + } + } + + const maybeFlush = async (drained: boolean): Promise => { + for (const ctx of contexts) { + // Loop: a full batch may leave more parked pieces behind it. + for (;;) { + const parked = db.parkedUploads(ctx.providerId) + const oldest = parked[0] + const reason = shouldFlush({ + batchSize: parked.length, + oldestParkedAtMs: oldest == null ? null : Date.parse(oldest.parkedAt), + nowMs: deps.now(), + assumedWindowMs: windowFor(ctx), + marginMs: marginFromConfirmations(observedCommitMs), + drained, + }) + if (reason == null) break + await flush(ctx, reason) + if (db.parkedUploads(ctx.providerId).length === parked.length) break // no progress; avoid spinning + } + } + await evictCommitted() + } + + // Reconcile add_unconfirmed leftovers from a previous run before uploading + // anything new: their outcome is unknown and a blind re-add would duplicate. + for (const ctx of contexts) { + await reconcileUnconfirmed(db, synapse, ctx, deps) + } + + // Main loop: store on the primary, fan out to secondaries, flush as batches + // and window timers demand. Sequential per piece: the upstream bandwidth is + // the bottleneck, and one in-flight store keeps the disk footprint bounded. + // In the pipeline an empty pending list means "wait for the packer", not + // "done": `waitForMore` resolves false only when the source is drained. + // A piece the provider disagreed with (commP mismatch) is recorded as a + // failed upload and drops out of this query, so it cannot re-store in a + // loop. + for (;;) { + const pending = db.subPiecesNeedingUpload(primary.providerId) + const next = pending[0] + if (next == null) { + if (opts.forceFlush?.() === true) { + // Downloads are blocked on the disk budget; commit whatever is + // parked now so eviction can free space, rather than holding the + // batch for the GC-window timer. + await maybeFlush(true) + } + if (opts.waitForMore != null && (await opts.waitForMore())) { + continue + } + break + } + if (next.carPath == null) { + // subPiecesNeedingUpload selects built rows, which always carry a CAR + // path; reaching this is a query bug. + throw new Error(`sub-piece ${next.subPieceCid} has no local CAR path`) + } + + const storeTimer = new Timer() + let stored: { size: number } + try { + stored = await storeCar(primary, deps, next.carPath, next.subPieceCid) + } catch (err) { + if (err instanceof CommPMismatchError) { + db.markUploadFailed(next.subPieceCid, primary.providerId, 'primary', err.message) + log(`error: ${err.message}`) + continue + } + throw err + } + storedBytes += stored.size + db.recordUploadParked(next.subPieceCid, primary.providerId, 'primary', primary.dataSetId) + log( + `parked ${next.subPieceCid} (${formatFileSize(stored.size)}) on primary ${primary.providerId} ` + + `in ${formatDuration(storeTimer.stop())}` + ) + + // Check the flush timers before the secondary fanout as well as after: a + // slow provider-to-provider pull must not age already-parked pieces past + // the GC window. + await maybeFlush(false) + for (const secondary of secondaries) { + await pullToSecondary(db, primary, secondary, next.subPieceCid) + } + + await maybeFlush(false) + } + + // Source drained. Flush whatever is parked, then retry what didn't land: + // collected pieces (GC'd before commit) and failed secondary pulls. A + // primary copy re-uploads from the staged CAR; a secondary copy re-pulls + // from the primary, which still holds the bytes. + await maybeFlush(true) + for (let attempt = 0; attempt < 3; attempt++) { + // A crash between the primary store and the secondary pull leaves a + // sub-piece with a live primary row and no row at all for the secondary; + // repair those alongside the recorded retry states, or the secondary + // copy would silently never exist. + const missingSecondaries = secondaries.flatMap((ctx) => + db.subPiecesMissingSecondary(primary.providerId, ctx.providerId).map((subPieceCid) => ({ ctx, subPieceCid })) + ) + const needsRetry = contexts.flatMap((ctx, i) => + ['collected' as const, ...(i > 0 ? ['failed' as const] : [])] + .flatMap((status) => db.uploadsByStatus(ctx.providerId, status)) + .map((u) => ({ ctx, u })) + ) + if (needsRetry.length === 0 && missingSecondaries.length === 0) break + log(`retrying ${needsRetry.length + missingSecondaries.length} piece(s) that did not land (attempt ${attempt + 1})`) + for (const { ctx, subPieceCid } of missingSecondaries) { + await pullToSecondary(db, primary, ctx, subPieceCid) + } + for (const { ctx, u } of needsRetry) { + if (u.role === 'secondary') { + await pullToSecondary(db, primary, ctx, u.subPieceCid) + continue + } + const sub = db.subPieceByCid(u.subPieceCid) + if (sub?.carPath == null) { + log(`error: collected ${u.subPieceCid} has no local CAR; cannot re-store`) + continue + } + try { + const stored = await storeCar(ctx, deps, sub.carPath, sub.subPieceCid) + storedBytes += stored.size + db.recordUploadParked(sub.subPieceCid, ctx.providerId, u.role, ctx.dataSetId) + } catch (err) { + if (err instanceof CommPMismatchError) { + db.markUploadFailed(sub.subPieceCid, ctx.providerId, u.role, err.message) + log(`error: ${err.message}`) + continue + } + throw err + } + } + await maybeFlush(true) + } + + await evictCommitted() + + const summary: DirectUploadSummary = { + network: synapse.chain.name, + providers: contexts.map((ctx, i) => { + const committed = db.uploadsByStatus(ctx.providerId, 'committed') + return { + providerId: ctx.providerId, + role: i === 0 ? ('primary' as const) : ('secondary' as const), + dataSetId: latestDataSetId(db, ctx), + committed: committed.length, + collected: db.uploadsByStatus(ctx.providerId, 'collected').length, + failed: db.uploadsByStatus(ctx.providerId, 'failed').length, + addUnconfirmed: db.uploadsByStatus(ctx.providerId, 'add_unconfirmed').length, + flushes: flushCounts.get(ctx.providerId) ?? 0, + assumedWindowMs: windowFor(ctx), + txHashes: [...new Set(committed.map((u) => u.txHash).filter((h): h is string => h != null))], + } + }), + storedBytes, + evictedCars: evictedPaths.size, + } + log(`direct upload finished in ${formatDuration(runTimer.stop())}: ${formatFileSize(storedBytes)} stored`) + return summary +} + +/** + * How old a hashless add_unconfirmed breadcrumb must be before an + * absent-on-chain verdict is trusted enough to re-queue the piece. Sized to + * outlast any realistic transaction confirmation window. + */ +export const HASHLESS_REQUEUE_AFTER_MS = 60 * 60_000 + +/** Thrown when the provider's commitment over the uploaded bytes disagrees with ours. */ +export class CommPMismatchError extends Error { + constructor(subPieceCid: string, providerId: string, got: string) { + super(`commP mismatch on provider ${providerId}: expected ${subPieceCid}, provider computed ${got}`) + this.name = 'CommPMismatchError' + } +} + +async function storeCar( + ctx: UploadContextLike, + deps: DirectUploadDeps, + carPath: string, + subPieceCid: string +): Promise<{ size: number }> { + const result = await ctx.store(deps.openCar(carPath), { pieceCid: CID.parse(subPieceCid) }) + // The committed CID must be the commitment over the bytes the provider + // actually holds; trusting the SDK to throw on divergence is not enough. + const got = String(result.pieceCid) + if (got !== subPieceCid) { + throw new CommPMismatchError(subPieceCid, ctx.providerId, got) + } + return { size: result.size } +} + +/** + * Resolve every add_unconfirmed row. The transaction receipt is checked first: + * piece presence alone cannot distinguish "commit never landed" from "commit + * landed but confirmation was missed": the provider keeps the bytes either + * way, and re-queueing a landed commit would add the piece twice. + */ +async function reconcileUnconfirmed( + db: MigrationDB, + synapse: Synapse, + ctx: UploadContextLike, + deps: DirectUploadDeps +): Promise { + for (const u of db.uploadsByStatus(ctx.providerId, 'add_unconfirmed')) { + if (u.txHash == null) { + // The crash landed between the transaction broadcast and the hash + // callback, so no receipt can be checked. The data set itself is the + // witness: a piece present on chain is committed; an absent one did + // not land. Without a known data set neither can be told apart, so the + // row stays unresolved and the run exits incomplete. + const dataSetId = u.dataSetId ?? ctx.dataSetId + if (dataSetId == null) { + log( + `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and no known data set ` + + `on provider ${ctx.providerId}; left add_unconfirmed for manual resolution` + ) + continue + } + const pieceId = await deps.dataSetPieceId(synapse, Number(dataSetId), u.subPieceCid) + if (pieceId != null) { + db.markUploadCommitted(u.subPieceCid, ctx.providerId, { dataSetId, pieceId, txHash: null }) + log(`resume: ${u.subPieceCid} found on chain in data set ${dataSetId} (piece ${pieceId}); marked committed`) + continue + } + // Absent on chain. A transaction the provider broadcast just before + // the crash could still land, and re-queueing while it can would add + // the piece twice, so the row only re-enters the flow once the + // breadcrumb is older than any realistic confirmation window. Younger + // rows stay unresolved and the run exits incomplete; a re-run later + // resolves them one way or the other. + const ageMs = deps.now() - Date.parse(u.updatedAt) + if (ageMs < HASHLESS_REQUEUE_AFTER_MS) { + log( + `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and is absent from ` + + `data set ${dataSetId}; too recent to rule out an in-flight transaction, left add_unconfirmed ` + + `(re-run after ${formatDuration(HASHLESS_REQUEUE_AFTER_MS - ageMs)})` + ) + continue + } + // Old enough that an in-flight transaction would have landed or died. + // Fall through to the presence check below so the piece re-parks or + // re-stores like any other unconfirmed attempt. + } + if (u.txHash != null && (await deps.txLanded(synapse, u.txHash))) { + // The commit landed; only local confirmation was missed. Resolve it from + // the canonical witness (the PiecesAdded event) rather than trusting + // any side channel. The row's own data set takes precedence: a resumed + // run may have opened a different context than the one that committed. + const dataSetId = u.dataSetId ?? ctx.dataSetId + if (dataSetId != null) { + const event = await deps.fetchAddPiecesEvent(synapse, Number(dataSetId), u.txHash) + const pieceIndex = event == null ? -1 : event.pieceCids.indexOf(u.subPieceCid) + if (event != null && pieceIndex >= 0) { + db.markUploadCommitted(u.subPieceCid, ctx.providerId, { + dataSetId, + pieceId: String(event.pieceIds[pieceIndex] ?? ''), + txHash: u.txHash, + }) + log( + `resume: ${u.subPieceCid} confirmed on chain via PiecesAdded (tx ${u.txHash}, ` + + `data set ${dataSetId}); marked committed` + ) + continue + } + } + log( + `resume: ${u.subPieceCid} has a LANDED addPieces tx ${u.txHash} on provider ${ctx.providerId} ` + + `but its PiecesAdded event could not be verified; leaving add_unconfirmed: check the data set ` + + `on the explorer before any manual retry (a blind re-add would duplicate the piece)` + ) + continue + } + if (await ctx.hasPiece(CID.parse(u.subPieceCid))) { + db.revertUploadsToParked([u.subPieceCid], ctx.providerId) + log(`resume: ${u.subPieceCid} still parked on provider ${ctx.providerId}; re-queued for commit`) + } else { + db.markUploadCollected(u.subPieceCid, ctx.providerId) + log(`resume: ${u.subPieceCid} gone from provider ${ctx.providerId}; will re-store`) + } + } +} + +/** Have one secondary pull a freshly parked piece from the primary. */ +async function pullToSecondary( + db: MigrationDB, + primary: UploadContextLike, + secondary: UploadContextLike, + subPieceCid: string +): Promise { + try { + // Curio authenticates the pull with the same EIP-712 authorization used + // for commit: a pull without it is rejected. + const extraData = await secondary.presignForCommit([{ pieceCid: CID.parse(subPieceCid) }]) + const pulled = await secondary.pull({ + pieces: [CID.parse(subPieceCid)], + from: (pieceCid) => primary.getPieceUrl(pieceCid), + extraData, + }) + if (pulled.status === 'complete') { + db.recordUploadParked(subPieceCid, secondary.providerId, 'secondary', secondary.dataSetId) + log(`parked ${subPieceCid} on secondary ${secondary.providerId} (pulled from primary)`) + } else { + db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', 'secondary pull failed') + log(`warn: secondary ${secondary.providerId} failed to pull ${subPieceCid}`) + } + } catch (err) { + db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', (err as Error).message) + log(`warn: secondary ${secondary.providerId} pull error for ${subPieceCid}: ${(err as Error).message}`) + } +} + +function latestDataSetId(db: MigrationDB, ctx: UploadContextLike): string | null { + const committed = db.uploadsByStatus(ctx.providerId, 'committed') + const last = committed[committed.length - 1] + return last != null ? last.dataSetId : ctx.dataSetId +} diff --git a/src/migrate/pdp-verifier.ts b/src/migrate/pdp-verifier.ts new file mode 100644 index 00000000..1fa0e24c --- /dev/null +++ b/src/migrate/pdp-verifier.ts @@ -0,0 +1,78 @@ +/** + * Read-only PDPVerifier access for reconciling a migrate run's on-chain + * state. The ABI and contract address come from the connected chain via + * `@filoz/synapse-core`. + */ + +import { pdp as PDP_ABI } from '@filoz/synapse-core/abis' +import { findPieceIdsByCid } from '@filoz/synapse-core/pdp-verifier' +import { from as pieceCidFrom } from '@filoz/synapse-core/piece' +import type { Synapse } from '@filoz/synapse-sdk' +import { type Hash, type Hex, parseEventLogs } from 'viem' + +/** + * The on-chain PiecesAdded event for a given data set, parsed from an + * AddPieces tx receipt. PDPVerifier emits one event per AddPieces call + * carrying parallel arrays of pieceIds + pieceCids (event + * `PiecesAdded(uint256 indexed setId, uint256[] pieceIds, struct Cids.Cid[] + * pieceCids)`). + */ +export interface AddPiecesEvent { + blockNumber: bigint + pieceIds: bigint[] + pieceCids: string[] +} + +/** + * The on-chain piece id of `pieceCid` in `dataSetId`, or null when the data + * set does not contain it. One contract read; used to resolve an + * add_unconfirmed row that never captured its transaction hash. + */ +export async function dataSetPieceId(synapse: Synapse, dataSetId: number, pieceCid: string): Promise { + const ids = await findPieceIdsByCid(synapse.client as never, { + dataSetId: BigInt(dataSetId), + pieceCid: pieceCidFrom(pieceCid), + }) + const first = ids[0] + return first == null ? null : String(first) +} + +/** Whether a transaction landed successfully on chain (receipt status success). */ +export async function txLanded(synapse: Synapse, txHash: string): Promise { + try { + const receipt = await synapse.client.getTransactionReceipt({ hash: txHash as Hash }) + return receipt.status === 'success' + } catch { + // No receipt yet (or the node dropped the tx): not landed. + return false + } +} + +/** + * Fetch and parse the PiecesAdded event matching `dataSetId` out of an + * AddPieces tx receipt. Returns null when the receipt carries no matching + * event (a reverted inner call leaves no PiecesAdded log even when the tx + * itself succeeded). + */ +export async function fetchAddPiecesEvent( + synapse: Synapse, + dataSetId: number, + txHash: string +): Promise { + const pdpAddress = synapse.chain.contracts.pdp.address + const receipt = await synapse.client.waitForTransactionReceipt({ hash: txHash as Hash }) + const events = parseEventLogs({ + abi: PDP_ABI, + eventName: 'PiecesAdded', + logs: receipt.logs, + }) + const target = BigInt(dataSetId) + const match = events.find((ev) => ev.address.toLowerCase() === pdpAddress.toLowerCase() && ev.args.setId === target) + if (match == null) return null + const pieceCids = match.args.pieceCids.map((p: { data: Hex }) => pieceCidFrom(p.data).toString()) + return { + blockNumber: receipt.blockNumber, + pieceIds: [...match.args.pieceIds], + pieceCids, + } +} diff --git a/src/test/unit/migrate-direct-upload-flow.test.ts b/src/test/unit/migrate-direct-upload-flow.test.ts new file mode 100644 index 00000000..53cd3227 --- /dev/null +++ b/src/test/unit/migrate-direct-upload-flow.test.ts @@ -0,0 +1,467 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import type { Synapse } from '@filoz/synapse-sdk' +import { describe, expect, it } from 'vitest' +import { MigrationDB } from '../../migrate/db.js' +import { + type DirectUploadDeps, + type DirectUploadOptions, + runDirectUpload, + type UploadContextLike, +} from '../../migrate/direct-upload.js' + +// Drives the real runDirectUpload control flow with fake providers, to lock in +// the direct-upload guarantees: store-then-batch-commit, the add_unconfirmed +// breadcrumb, GC detection lowering the window and re-storing only what is +// actually gone, commP verification of what the provider received, and CAR +// eviction once the primary copy is committed. + +// Real PieceCIDs so CID.parse succeeds. +const P1 = 'bafkzcibf3ck4uais4fgennh4hbfx5z3i6hue4xgq2cdeamtus4hjbsrjs5lf2azbxmsa' +const P2 = 'bafkzcibewpkqwewyhz3yxutlxbpt2nkb6si5qilg4qqtzzij32uw7ammsc73a4wkgi' + +const fakeSynapse = { chain: { name: 'calibration' } } as unknown as Synapse +const OPTS: DirectUploadOptions = { synapse: fakeSynapse, copies: 2 } +const SCOPE = 'calibration:0xabc' + +async function dbAt(name: string) { + const dir = await mkdtemp(join(tmpdir(), `fp-${name}-`)) + return { dir, db: new MigrationDB(join(dir, 'migrate.db'), SCOPE) } +} + +function seedBuilt(db: MigrationDB, subPieceCid: string, carPath: string) { + const src = `src-${subPieceCid.slice(-8)}` + db.addCids([src]) + db.recordPieceSuccess(src, { + pieceCid: subPieceCid, + rawSize: 1024, + gateway: 'g', + url: `https://gw/ipfs/${src}?format=car`, + memberCarPath: `${carPath}.member`, + memberSha256: 'sha', + }) + db.recordBuiltSubPiece({ + subPieceCid, + assembledCarLength: 1024, + targetSizeBytes: 1024, + carPath, + assembledSha256: 'sha', + members: [{ cid: src, sha256: 'sha', rawSize: 1024 }], + }) +} + +interface FakeBehavior { + /** Throw on the numbered commit call (1-based) on the given provider. */ + failCommit?: { providerId: string; call: number; message: string } + /** Per-CID presence answers for hasPiece during re-verify. */ + present?: (cid: string) => boolean + pullFails?: boolean + /** Store answers a different commitment than requested (commP mismatch). */ + storeReturns?: (pieceCid: string) => string + /** Receipt answers for add_unconfirmed reconciliation. Default: not landed. */ + txLanded?: (txHash: string) => boolean + /** PiecesAdded event answers for landed-tx resolution. Default: none found. */ + addPiecesEvent?: (dataSetId: number, txHash: string) => { pieceIds: bigint[]; pieceCids: string[] } | null + /** On-chain piece-id answers for hashless reconciliation. Default: absent. */ + dataSetPieceId?: (dataSetId: number, pieceCid: string) => string | null + /** Pre-resolved data set id for every context. Default null (created lazily). */ + ctxDataSetId?: string + /** Offset added to the fake clock, for aging add_unconfirmed breadcrumbs. */ + nowOffsetMs?: number +} + +function fakeDeps(b: FakeBehavior = {}) { + const calls = { store: [] as string[], pull: 0, commit: new Map() } + const evicted: string[] = [] + const mkCtx = (providerId: string): UploadContextLike => ({ + providerId, + serviceURL: `fake://${providerId}`, + dataSetId: b.ctxDataSetId ?? null, + async store(_data, options) { + const requested = String(options.pieceCid) + calls.store.push(`${providerId}:${requested}`) + const answered = b.storeReturns ? b.storeReturns(requested) : requested + return { pieceCid: answered, size: 1024 } + }, + async presignForCommit() { + return '0xfake' + }, + async pull(options) { + calls.pull++ + const status = b.pullFails ? ('failed' as const) : ('complete' as const) + return { status, pieces: options.pieces.map((p) => ({ pieceCid: p, status })) } + }, + async commit(options) { + const n = (calls.commit.get(providerId) ?? 0) + 1 + calls.commit.set(providerId, n) + const f = b.failCommit + if (f != null && f.providerId === providerId && f.call === n) { + // Simulate a commit that was submitted (tx hash recorded) but then + // failed at confirmation: the shape of the add_unconfirmed hazard. + options.onSubmitted?.(`0xtx-${providerId}-${n}`) + throw new Error(f.message) + } + return { + txHash: `0xtx-${providerId}-${n}`, + pieceIds: options.pieces.map((_, i) => BigInt(i)), + dataSetId: 7n, + } + }, + getPieceUrl: (pieceCid) => `fake://${providerId}/piece/${String(pieceCid)}`, + hasPiece: async (pieceCid) => (b.present ? b.present(String(pieceCid)) : true), + }) + const deps: DirectUploadDeps = { + async setup() { + return { contexts: [mkCtx('p1'), mkCtx('p2')] } + }, + now: () => Date.now() + (b.nowOffsetMs ?? 0), + openCar: () => new Uint8Array(8), + evictCar: async (path) => { + evicted.push(path) + }, + txLanded: async (_synapse, txHash) => (b.txLanded ? b.txLanded(txHash) : false), + fetchAddPiecesEvent: async (_synapse, dataSetId, txHash) => { + const event = b.addPiecesEvent ? b.addPiecesEvent(dataSetId, txHash) : null + return event == null ? null : { ...event, blockNumber: 1n } + }, + dataSetPieceId: async (_synapse, dataSetId, pieceCid) => + b.dataSetPieceId ? b.dataSetPieceId(dataSetId, pieceCid) : null, + } + return { deps, calls, evicted } +} + +describe('runDirectUpload', () => { + it('happy path: stores on primary, pulls to secondary, drained flush commits both, evicts CARs', async () => { + const { dir, db } = await dbAt('du-happy') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + seedBuilt(db, P2, join(dir, 'b.car')) + const { deps, calls, evicted } = fakeDeps() + const summary = await runDirectUpload(db, OPTS, deps) + + expect(calls.store).toEqual([`p1:${P1}`, `p1:${P2}`]) + expect(calls.pull).toBe(2) + // One drained flush per provider, both pieces in one batch. + expect(calls.commit.get('p1')).toBe(1) + expect(calls.commit.get('p2')).toBe(1) + for (const provider of ['p1', 'p2']) { + const committed = db.uploadsByStatus(provider, 'committed') + expect(committed.map((u) => u.subPieceCid).sort()).toEqual([P2, P1].sort()) + for (const u of committed) expect(u.dataSetId).toBe('7') + } + expect(evicted).toHaveLength(2) + expect(summary.providers[0]?.committed).toBe(2) + expect(summary.providers[0]?.role).toBe('primary') + expect(summary.providers[0]?.addUnconfirmed).toBe(0) + // The summary must surface the addPieces transactions behind the commits. + expect(summary.providers[0]?.txHashes).toEqual(['0xtx-p1-1']) + expect(summary.providers[1]?.txHashes).toEqual(['0xtx-p2-1']) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('secondary pull failure leaves the secondary failed but frees the CAR once the primary commits', async () => { + const { dir, db } = await dbAt('du-pullfail') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + const { deps, evicted } = fakeDeps({ pullFails: true }) + const summary = await runDirectUpload(db, OPTS, deps) + + expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1) + expect(db.uploadsByStatus('p2', 'failed')).toHaveLength(1) + // A secondary retry pulls provider-to-provider from the committed + // primary; it never needs the local file, so the CAR is evicted. + expect(evicted).toHaveLength(1) + // The failed copy still marks the run incomplete for the exit code. + expect(summary.providers[1]?.failed).toBe(1) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('a commP mismatch from store() fails the piece instead of committing the wrong bytes', async () => { + const { dir, db } = await dbAt('du-commp') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + const { deps, calls, evicted } = fakeDeps({ storeReturns: () => P2 }) + await runDirectUpload(db, OPTS, deps) + + const failed = db.uploadsByStatus('p1', 'failed') + expect(failed).toHaveLength(1) + expect(failed[0]?.error).toMatch(/commP mismatch/) + expect(calls.commit.get('p1') ?? 0).toBe(0) + expect(evicted).toHaveLength(0) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('repairs a missing secondary copy left by a crash between store and pull', async () => { + const { dir, db } = await dbAt('du-repair') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + // The crash shape: the primary parked but the secondary pull never + // started, so the secondary has no row at all. + db.recordUploadParked(P1, 'p1', 'primary', null) + const { deps, calls } = fakeDeps() + await runDirectUpload(db, OPTS, deps) + + // No re-store on the primary; the secondary copy was pulled and both + // copies committed. + expect(calls.store).toHaveLength(0) + expect(calls.pull).toBe(1) + expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1) + expect(db.uploadsByStatus('p2', 'committed')).toHaveLength(1) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('GC rejection: lowers the provider window, re-stores only the collected piece, keeps the rest parked', async () => { + const { dir, db } = await dbAt('du-gc') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + seedBuilt(db, P2, join(dir, 'b.car')) + const { deps, calls } = fakeDeps({ + failCommit: { + providerId: 'p1', + call: 1, + message: `Failed to process request: subPiece CID ${P1} not found or does not belong to service svc`, + }, + present: (cid) => cid !== P1, + }) + await runDirectUpload(db, OPTS, deps) + + // P1 was re-stored on the primary after being collected; P2 was not. + expect(calls.store.filter((s) => s === `p1:${P1}`)).toHaveLength(2) + expect(calls.store.filter((s) => s === `p1:${P2}`)).toHaveLength(1) + // Both pieces end up committed via the retry flush. + expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(2) + // The window guess dropped below the default for the flaky provider only. + const defaultMs = 60 * 60_000 + expect(db.providerWindowMs('p1', defaultMs)).toBeLessThan(defaultMs) + expect(db.providerWindowMs('p2', defaultMs)).toBe(defaultMs) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('a landed-but-unconfirmed tx is never re-queued: no blind re-add', async () => { + const { dir, db } = await dbAt('du-landed') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + const first = fakeDeps({ + failCommit: { providerId: 'p1', call: 1, message: 'confirmation poll timed out' }, + }) + await runDirectUpload(db, OPTS, first.deps) + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + + // Second run: the tx actually landed on chain even though confirmation + // was missed. The reconciliation must leave the row alone: no store, + // no commit: and the summary must count it. + const second = fakeDeps({ txLanded: () => true }) + const summary = await runDirectUpload(db, OPTS, second.deps) + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + expect(second.calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0) + expect(second.calls.commit.get('p1') ?? 0).toBe(0) + expect(summary.providers[0]?.addUnconfirmed).toBe(1) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('a landed tx resolves from the PiecesAdded event using the data set recorded on the row', async () => { + const { dir, db } = await dbAt('du-landed-event') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + const first = fakeDeps({ + ctxDataSetId: '7', + failCommit: { providerId: 'p1', call: 1, message: 'confirmation poll timed out' }, + }) + await runDirectUpload(db, OPTS, first.deps) + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + + // Second run opens fresh contexts with no resolved data set; the row's + // own data_set_id (recorded when the piece parked) must drive the + // event lookup. + const second = fakeDeps({ + txLanded: () => true, + addPiecesEvent: (dataSetId) => (dataSetId === 7 ? { pieceIds: [42n], pieceCids: [P1] } : null), + }) + await runDirectUpload(db, OPTS, second.deps) + const committed = db.uploadsByStatus('p1', 'committed') + expect(committed).toHaveLength(1) + expect(committed[0]?.pieceId).toBe('42') + expect(committed[0]?.dataSetId).toBe('7') + // Resolved from the chain, not re-executed. + expect(second.calls.commit.get('p1') ?? 0).toBe(0) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('non-GC commit failure leaves the batch add_unconfirmed with no stale tx hash after re-park', async () => { + const { dir, db } = await dbAt('du-unconfirmed') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + const first = fakeDeps({ + failCommit: { providerId: 'p1', call: 1, message: 'insufficient funds' }, + }) + await runDirectUpload(db, OPTS, first.deps) + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + + // Second run: the piece is still parked on the provider, so the resume + // reconciliation re-queues it and the commit lands, without re-storing + // and with the failed attempt's tx hash replaced by the real one. + const second = fakeDeps() + await runDirectUpload(db, OPTS, second.deps) + const committed = db.uploadsByStatus('p1', 'committed') + expect(committed).toHaveLength(1) + expect(committed[0]?.txHash).toBe('0xtx-p1-1') + expect(second.calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('resolves a hashless add_unconfirmed row from the data set itself when the piece is on chain', async () => { + const { dir, db } = await dbAt('du-hashless-committed') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + // The crash shape: the breadcrumb was written and the commit may have + // broadcast, but the process died before the tx hash callback. + db.recordUploadParked(P1, 'p1', 'primary', '7') + db.markUploadsAddUnconfirmed([P1], 'p1') + + const { deps, calls } = fakeDeps({ dataSetPieceId: (dataSetId) => (dataSetId === 7 ? '42' : null) }) + await runDirectUpload(db, OPTS, deps) + + const committed = db.uploadsByStatus('p1', 'committed') + expect(committed).toHaveLength(1) + expect(committed[0]?.pieceId).toBe('42') + // Resolved from the chain: no re-store, no second commit for it. + expect(calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('holds a fresh hashless add_unconfirmed row even when absent from the data set', async () => { + const { dir, db } = await dbAt('du-hashless-fresh') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + db.recordUploadParked(P1, 'p1', 'primary', '7') + db.markUploadsAddUnconfirmed([P1], 'p1') + + // A just-written breadcrumb cannot rule out a transaction still in + // flight; the row must stay unresolved this run. + const { deps, calls } = fakeDeps() + const summary = await runDirectUpload(db, OPTS, deps) + + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + expect(calls.commit.get('p1') ?? 0).toBe(0) + expect(summary.providers[0]?.addUnconfirmed).toBe(1) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('re-parks a hashless add_unconfirmed row absent from the data set once it has aged out', async () => { + const { dir, db } = await dbAt('du-hashless-aged') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + db.recordUploadParked(P1, 'p1', 'primary', '7') + db.markUploadsAddUnconfirmed([P1], 'p1') + + // Two hours later, an in-flight transaction would have landed or died; + // absent on chain and still parked on the provider means re-queue, and + // the commit lands without re-storing. + const { deps, calls } = fakeDeps({ nowOffsetMs: 2 * 60 * 60_000 }) + await runDirectUpload(db, OPTS, deps) + + expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1) + expect(calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0) + expect(calls.commit.get('p1')).toBe(1) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('leaves a hashless add_unconfirmed row alone when no data set is known', async () => { + const { dir, db } = await dbAt('du-hashless-unknown') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + db.recordUploadParked(P1, 'p1', 'primary', null) + db.markUploadsAddUnconfirmed([P1], 'p1') + + const { deps, calls } = fakeDeps() + const summary = await runDirectUpload(db, OPTS, deps) + + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + expect(calls.commit.get('p1') ?? 0).toBe(0) + expect(summary.providers[0]?.addUnconfirmed).toBe(1) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('refuses to rebuild a staged piece while it has live upload state', async () => { + const { dir, db } = await dbAt('du-rebuild-guard') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + db.recordUploadParked(P1, 'p1', 'primary', '7') + db.markUploadsAddUnconfirmed([P1], 'p1') + + // An unresolved breadcrumb must survive any rebuild attempt: deleting + // it and re-adding the source CIDs could duplicate the piece on chain. + expect(() => db.deleteSubPieceForRebuild(P1)).toThrow(/live upload/) + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + + // Once reconciliation resolves the row to a terminal retry state, the + // rebuild goes through and frees the members. + db.markUploadCollected(P1, 'p1') + const members = db.deleteSubPieceForRebuild(P1) + expect(members).toHaveLength(1) + expect(db.subPieceByCid(P1)).toBeNull() + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('scopes state: another network/owner scope sees none of the rows', async () => { + const { dir, db } = await dbAt('du-scope') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + const { deps } = fakeDeps() + await runDirectUpload(db, OPTS, deps) + expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1) + + const other = new MigrationDB(db.path, 'mainnet:0xdef') + try { + expect(other.subPiecesNeedingUpload('p1')).toHaveLength(0) + expect(other.uploadsByStatus('p1', 'committed')).toHaveLength(0) + expect(other.counts().total).toBe(0) + } finally { + other.close() + } + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) +}) From fa1d911a311358ab1149d016b3353d291d45c4b5 Mon Sep 17 00:00:00 2001 From: Russell Dempsey <1173416+SgtPooki@users.noreply.github.com> Date: Thu, 20 Aug 2026 13:50:21 -0400 Subject: [PATCH 2/3] fix: close the double-add windows in reconciliation An add_unconfirmed row whose tx had no receipt fell straight through to the presence check and re-parked, so the next flush could issue a second addPieces while the original tx was still in the mempool. The same age gate the hashless path uses now holds the row until the breadcrumb outlives any realistic confirmation window (the constant is renamed UNCONFIRMED_REQUEUE_AFTER_MS to cover both paths). txLanded returned false on any RPC error, so one 5xx or rate limit looked like a dead transaction and fed that re-queue. Only viem's TransactionReceiptNotFoundError maps to false now; other errors abort the pass. fetchAddPiecesEvent no longer requires a known data set id: a crash on the first commit of a new data set records none, and the event's own setId is the recovery path. A successful commit writes the resolved id back onto the context so later parks carry it. hasPiece treats only the SP's not-found answer as absence, and data set ids stay strings end to end instead of passing through Number. --- src/migrate/direct-upload.ts | 75 ++++++++++++------- src/migrate/pdp-verifier.ts | 56 +++++++++----- .../unit/migrate-direct-upload-flow.test.ts | 73 +++++++++++++++--- 3 files changed, 150 insertions(+), 54 deletions(-) diff --git a/src/migrate/direct-upload.ts b/src/migrate/direct-upload.ts index 0987fada..de6a11fb 100644 --- a/src/migrate/direct-upload.ts +++ b/src/migrate/direct-upload.ts @@ -21,6 +21,7 @@ import { createReadStream } from 'node:fs' import { unlink } from 'node:fs/promises' import { Readable } from 'node:stream' +import { FindPieceError } from '@filoz/synapse-core/errors' import { findPiece } from '@filoz/synapse-core/sp' import type { Synapse } from '@filoz/synapse-sdk' import { CID } from 'multiformats/cid' @@ -116,9 +117,9 @@ export interface DirectUploadDeps { /** Whether a transaction landed successfully on chain (receipt status 1). */ txLanded(synapse: Synapse, txHash: string): Promise /** Canonical on-chain witness for a landed addPieces (see pdp-verifier). */ - fetchAddPiecesEvent(synapse: Synapse, dataSetId: number, txHash: string): Promise + fetchAddPiecesEvent(synapse: Synapse, txHash: string, dataSetId: string | null): Promise /** On-chain piece id lookup for a data set, or null when absent (see pdp-verifier). */ - dataSetPieceId(synapse: Synapse, dataSetId: number, pieceCid: string): Promise + dataSetPieceId(synapse: Synapse, dataSetId: string, pieceCid: string): Promise } export const defaultDirectUploadDeps: DirectUploadDeps = { @@ -150,8 +151,12 @@ export const defaultDirectUploadDeps: DirectUploadDeps = { try { await findPiece({ serviceURL, pieceCid: pieceCid as never, retryCount: 0 }) return true - } catch { - return false + } catch (err) { + // False feeds "collected" transitions that trigger a full + // re-store, so only the SP's own not-found answer counts. + // Timeouts and transport errors propagate and abort the pass. + if (err instanceof FindPieceError && !err.message.includes('Timeout')) return false + throw err } }, } @@ -245,6 +250,10 @@ export async function runDirectUpload( txHash: result.txHash, }) }) + // A context created without a data set learns its id at the first + // commit; recording it here means later parks and resumes carry the + // set id on the row instead of depending on context re-resolution. + ctx.dataSetId = String(result.dataSetId) log(`committed ${batch.length} piece(s) on provider ${ctx.providerId} (data set ${result.dataSetId})`) } catch (err) { const message = (err as Error).message ?? String(err) @@ -459,11 +468,13 @@ export async function runDirectUpload( } /** - * How old a hashless add_unconfirmed breadcrumb must be before an - * absent-on-chain verdict is trusted enough to re-queue the piece. Sized to - * outlast any realistic transaction confirmation window. + * How old an add_unconfirmed breadcrumb must be before its piece re-enters + * the flow: for a hashless row, before an absent-on-chain verdict is + * trusted; for a row whose tx has no receipt, before the tx is presumed + * dead. Sized to outlast any realistic confirmation window, because the + * failure mode on both paths is a duplicate on-chain add. */ -export const HASHLESS_REQUEUE_AFTER_MS = 60 * 60_000 +export const UNCONFIRMED_REQUEUE_AFTER_MS = 60 * 60_000 /** Thrown when the provider's commitment over the uploaded bytes disagrees with ours. */ export class CommPMismatchError extends Error { @@ -516,7 +527,7 @@ async function reconcileUnconfirmed( ) continue } - const pieceId = await deps.dataSetPieceId(synapse, Number(dataSetId), u.subPieceCid) + const pieceId = await deps.dataSetPieceId(synapse, dataSetId, u.subPieceCid) if (pieceId != null) { db.markUploadCommitted(u.subPieceCid, ctx.providerId, { dataSetId, pieceId, txHash: null }) log(`resume: ${u.subPieceCid} found on chain in data set ${dataSetId} (piece ${pieceId}); marked committed`) @@ -529,11 +540,11 @@ async function reconcileUnconfirmed( // rows stay unresolved and the run exits incomplete; a re-run later // resolves them one way or the other. const ageMs = deps.now() - Date.parse(u.updatedAt) - if (ageMs < HASHLESS_REQUEUE_AFTER_MS) { + if (ageMs < UNCONFIRMED_REQUEUE_AFTER_MS) { log( `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and is absent from ` + `data set ${dataSetId}; too recent to rule out an in-flight transaction, left add_unconfirmed ` + - `(re-run after ${formatDuration(HASHLESS_REQUEUE_AFTER_MS - ageMs)})` + `(re-run after ${formatDuration(UNCONFIRMED_REQUEUE_AFTER_MS - ageMs)})` ) continue } @@ -541,16 +552,17 @@ async function reconcileUnconfirmed( // Fall through to the presence check below so the piece re-parks or // re-stores like any other unconfirmed attempt. } - if (u.txHash != null && (await deps.txLanded(synapse, u.txHash))) { - // The commit landed; only local confirmation was missed. Resolve it from - // the canonical witness (the PiecesAdded event) rather than trusting - // any side channel. The row's own data set takes precedence: a resumed - // run may have opened a different context than the one that committed. - const dataSetId = u.dataSetId ?? ctx.dataSetId - if (dataSetId != null) { - const event = await deps.fetchAddPiecesEvent(synapse, Number(dataSetId), u.txHash) + if (u.txHash != null) { + if (await deps.txLanded(synapse, u.txHash)) { + // The commit landed; only local confirmation was missed. Resolve it + // from the canonical witness (the PiecesAdded event) rather than + // trusting any side channel. The row's own data set takes precedence + // when known; a first-commit crash on a new data set has none, and + // then the event's own setId is the recovery path. + const event = await deps.fetchAddPiecesEvent(synapse, u.txHash, u.dataSetId ?? ctx.dataSetId) const pieceIndex = event == null ? -1 : event.pieceCids.indexOf(u.subPieceCid) if (event != null && pieceIndex >= 0) { + const dataSetId = String(event.dataSetId) db.markUploadCommitted(u.subPieceCid, ctx.providerId, { dataSetId, pieceId: String(event.pieceIds[pieceIndex] ?? ''), @@ -562,13 +574,26 @@ async function reconcileUnconfirmed( ) continue } + log( + `resume: ${u.subPieceCid} has a LANDED addPieces tx ${u.txHash} on provider ${ctx.providerId} ` + + `but its PiecesAdded event could not be verified; leaving add_unconfirmed: check the data set ` + + `on the explorer before any manual retry (a blind re-add would duplicate the piece)` + ) + continue + } + // Not landed is not dead: the tx can still be in the mempool, and + // re-parking now would let the next flush double-add once it lands. + // Same age gate as the hashless path: only a breadcrumb older than any + // realistic confirmation window re-enters the flow. + const ageMs = deps.now() - Date.parse(u.updatedAt) + if (ageMs < UNCONFIRMED_REQUEUE_AFTER_MS) { + log( + `resume: ${u.subPieceCid} has an unlanded addPieces tx ${u.txHash} on provider ${ctx.providerId}; ` + + `too recent to rule out an in-flight transaction, left add_unconfirmed ` + + `(re-run after ${formatDuration(UNCONFIRMED_REQUEUE_AFTER_MS - ageMs)})` + ) + continue } - log( - `resume: ${u.subPieceCid} has a LANDED addPieces tx ${u.txHash} on provider ${ctx.providerId} ` + - `but its PiecesAdded event could not be verified; leaving add_unconfirmed: check the data set ` + - `on the explorer before any manual retry (a blind re-add would duplicate the piece)` - ) - continue } if (await ctx.hasPiece(CID.parse(u.subPieceCid))) { db.revertUploadsToParked([u.subPieceCid], ctx.providerId) diff --git a/src/migrate/pdp-verifier.ts b/src/migrate/pdp-verifier.ts index 1fa0e24c..716c6a13 100644 --- a/src/migrate/pdp-verifier.ts +++ b/src/migrate/pdp-verifier.ts @@ -8,16 +8,18 @@ import { pdp as PDP_ABI } from '@filoz/synapse-core/abis' import { findPieceIdsByCid } from '@filoz/synapse-core/pdp-verifier' import { from as pieceCidFrom } from '@filoz/synapse-core/piece' import type { Synapse } from '@filoz/synapse-sdk' -import { type Hash, type Hex, parseEventLogs } from 'viem' +import { type Hash, type Hex, parseEventLogs, TransactionReceiptNotFoundError } from 'viem' /** - * The on-chain PiecesAdded event for a given data set, parsed from an - * AddPieces tx receipt. PDPVerifier emits one event per AddPieces call - * carrying parallel arrays of pieceIds + pieceCids (event - * `PiecesAdded(uint256 indexed setId, uint256[] pieceIds, struct Cids.Cid[] - * pieceCids)`). + * The on-chain PiecesAdded event parsed from an AddPieces tx receipt. + * PDPVerifier emits one event per AddPieces call carrying parallel arrays of + * pieceIds + pieceCids (event `PiecesAdded(uint256 indexed setId, uint256[] + * pieceIds, struct Cids.Cid[] pieceCids)`); `dataSetId` is the event's own + * setId, so a caller that lost track of its data set can recover it from + * the receipt. */ export interface AddPiecesEvent { + dataSetId: bigint blockNumber: bigint pieceIds: bigint[] pieceCids: string[] @@ -26,9 +28,11 @@ export interface AddPiecesEvent { /** * The on-chain piece id of `pieceCid` in `dataSetId`, or null when the data * set does not contain it. One contract read; used to resolve an - * add_unconfirmed row that never captured its transaction hash. + * add_unconfirmed row that never captured its transaction hash. `dataSetId` + * stays a string end to end: on-chain set ids are uint256 and `Number` + * would silently truncate. */ -export async function dataSetPieceId(synapse: Synapse, dataSetId: number, pieceCid: string): Promise { +export async function dataSetPieceId(synapse: Synapse, dataSetId: string, pieceCid: string): Promise { const ids = await findPieceIdsByCid(synapse.client as never, { dataSetId: BigInt(dataSetId), pieceCid: pieceCidFrom(pieceCid), @@ -37,27 +41,36 @@ export async function dataSetPieceId(synapse: Synapse, dataSetId: number, pieceC return first == null ? null : String(first) } -/** Whether a transaction landed successfully on chain (receipt status success). */ +/** + * Whether a transaction landed successfully on chain (receipt status + * success). Only a definitive missing receipt maps to false; any other + * error (RPC 5xx, rate limit, timeout) propagates, because callers make + * re-add decisions on a false and a transient blip must not look like a + * dead transaction. + */ export async function txLanded(synapse: Synapse, txHash: string): Promise { try { const receipt = await synapse.client.getTransactionReceipt({ hash: txHash as Hash }) return receipt.status === 'success' - } catch { - // No receipt yet (or the node dropped the tx): not landed. - return false + } catch (err) { + if (err instanceof TransactionReceiptNotFoundError) return false + throw err } } /** - * Fetch and parse the PiecesAdded event matching `dataSetId` out of an - * AddPieces tx receipt. Returns null when the receipt carries no matching - * event (a reverted inner call leaves no PiecesAdded log even when the tx - * itself succeeded). + * Fetch and parse the PiecesAdded event out of an AddPieces tx receipt. + * When `dataSetId` is known it must match the event's setId; when null (a + * crash before the first commit of a new data set recorded one) any + * PDPVerifier-emitted event in the receipt matches and its setId is + * returned. Returns null when the receipt carries no matching event (a + * reverted inner call leaves no PiecesAdded log even when the tx itself + * succeeded). */ export async function fetchAddPiecesEvent( synapse: Synapse, - dataSetId: number, - txHash: string + txHash: string, + dataSetId: string | null ): Promise { const pdpAddress = synapse.chain.contracts.pdp.address const receipt = await synapse.client.waitForTransactionReceipt({ hash: txHash as Hash }) @@ -66,11 +79,14 @@ export async function fetchAddPiecesEvent( eventName: 'PiecesAdded', logs: receipt.logs, }) - const target = BigInt(dataSetId) - const match = events.find((ev) => ev.address.toLowerCase() === pdpAddress.toLowerCase() && ev.args.setId === target) + const target = dataSetId == null ? null : BigInt(dataSetId) + const match = events.find( + (ev) => ev.address.toLowerCase() === pdpAddress.toLowerCase() && (target == null || ev.args.setId === target) + ) if (match == null) return null const pieceCids = match.args.pieceCids.map((p: { data: Hex }) => pieceCidFrom(p.data).toString()) return { + dataSetId: match.args.setId, blockNumber: receipt.blockNumber, pieceIds: [...match.args.pieceIds], pieceCids, diff --git a/src/test/unit/migrate-direct-upload-flow.test.ts b/src/test/unit/migrate-direct-upload-flow.test.ts index 53cd3227..cb5f9bf5 100644 --- a/src/test/unit/migrate-direct-upload-flow.test.ts +++ b/src/test/unit/migrate-direct-upload-flow.test.ts @@ -62,9 +62,12 @@ interface FakeBehavior { /** Receipt answers for add_unconfirmed reconciliation. Default: not landed. */ txLanded?: (txHash: string) => boolean /** PiecesAdded event answers for landed-tx resolution. Default: none found. */ - addPiecesEvent?: (dataSetId: number, txHash: string) => { pieceIds: bigint[]; pieceCids: string[] } | null + addPiecesEvent?: ( + dataSetId: string | null, + txHash: string + ) => { dataSetId: bigint; pieceIds: bigint[]; pieceCids: string[] } | null /** On-chain piece-id answers for hashless reconciliation. Default: absent. */ - dataSetPieceId?: (dataSetId: number, pieceCid: string) => string | null + dataSetPieceId?: (dataSetId: string, pieceCid: string) => string | null /** Pre-resolved data set id for every context. Default null (created lazily). */ ctxDataSetId?: string /** Offset added to the fake clock, for aging add_unconfirmed breadcrumbs. */ @@ -121,7 +124,7 @@ function fakeDeps(b: FakeBehavior = {}) { evicted.push(path) }, txLanded: async (_synapse, txHash) => (b.txLanded ? b.txLanded(txHash) : false), - fetchAddPiecesEvent: async (_synapse, dataSetId, txHash) => { + fetchAddPiecesEvent: async (_synapse, txHash, dataSetId) => { const event = b.addPiecesEvent ? b.addPiecesEvent(dataSetId, txHash) : null return event == null ? null : { ...event, blockNumber: 1n } }, @@ -294,7 +297,8 @@ describe('runDirectUpload', () => { // event lookup. const second = fakeDeps({ txLanded: () => true, - addPiecesEvent: (dataSetId) => (dataSetId === 7 ? { pieceIds: [42n], pieceCids: [P1] } : null), + addPiecesEvent: (dataSetId) => + dataSetId === '7' ? { dataSetId: 7n, pieceIds: [42n], pieceCids: [P1] } : null, }) await runDirectUpload(db, OPTS, second.deps) const committed = db.uploadsByStatus('p1', 'committed') @@ -319,10 +323,13 @@ describe('runDirectUpload', () => { await runDirectUpload(db, OPTS, first.deps) expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) - // Second run: the piece is still parked on the provider, so the resume - // reconciliation re-queues it and the commit lands, without re-storing - // and with the failed attempt's tx hash replaced by the real one. - const second = fakeDeps() + // Second run, past the requeue window (a younger row with a captured + // tx hash is held: the tx could still land and re-adding would + // double-add). The piece is still parked on the provider, so the + // resume reconciliation re-queues it and the commit lands, without + // re-storing and with the failed attempt's tx hash replaced by the + // real one. + const second = fakeDeps({ nowOffsetMs: 61 * 60_000 }) await runDirectUpload(db, OPTS, second.deps) const committed = db.uploadsByStatus('p1', 'committed') expect(committed).toHaveLength(1) @@ -343,7 +350,7 @@ describe('runDirectUpload', () => { db.recordUploadParked(P1, 'p1', 'primary', '7') db.markUploadsAddUnconfirmed([P1], 'p1') - const { deps, calls } = fakeDeps({ dataSetPieceId: (dataSetId) => (dataSetId === 7 ? '42' : null) }) + const { deps, calls } = fakeDeps({ dataSetPieceId: (dataSetId) => (dataSetId === '7' ? '42' : null) }) await runDirectUpload(db, OPTS, deps) const committed = db.uploadsByStatus('p1', 'committed') @@ -357,6 +364,54 @@ describe('runDirectUpload', () => { } }) + it('holds a fresh add_unconfirmed row whose tx has no receipt yet', async () => { + const { dir, db } = await dbAt('du-pending-tx') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + db.recordUploadParked(P1, 'p1', 'primary', '7') + db.markUploadsAddUnconfirmed([P1], 'p1') + db.markUploadTxSubmitted([P1], 'p1', '0xpending') + + // txLanded false = no receipt. The tx can still be in the mempool, so + // the row must not re-park (a second commit would double-add). + const { deps, calls } = fakeDeps({ txLanded: () => false }) + await runDirectUpload(db, OPTS, deps) + + expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1) + expect(calls.commit.get('p1') ?? 0).toBe(0) + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + + it('recovers the data set id from the PiecesAdded event when none was recorded', async () => { + const { dir, db } = await dbAt('du-event-setid') + try { + seedBuilt(db, P1, join(dir, 'a.car')) + // First-commit crash shape on a new data set: parked before any set id + // existed, tx hash captured, no data_set_id anywhere. + db.recordUploadParked(P1, 'p1', 'primary', null) + db.markUploadsAddUnconfirmed([P1], 'p1') + db.markUploadTxSubmitted([P1], 'p1', '0xfirst') + + const { deps } = fakeDeps({ + txLanded: () => true, + addPiecesEvent: (dataSetId) => + dataSetId == null ? { dataSetId: 9n, pieceIds: [5n], pieceCids: [P1] } : null, + }) + await runDirectUpload(db, OPTS, deps) + + const committed = db.uploadsByStatus('p1', 'committed') + expect(committed).toHaveLength(1) + expect(committed[0]?.dataSetId).toBe('9') + expect(committed[0]?.pieceId).toBe('5') + } finally { + db.close() + await rm(dir, { recursive: true, force: true }) + } + }) + it('holds a fresh hashless add_unconfirmed row even when absent from the data set', async () => { const { dir, db } = await dbAt('du-hashless-fresh') try { From 88ff29051ffcb2c935e5f414de7195df8f032b54 Mon Sep 17 00:00:00 2001 From: Russell Dempsey <1173416+SgtPooki@users.noreply.github.com> Date: Wed, 26 Aug 2026 14:41:05 -0400 Subject: [PATCH 3/3] refactor: route direct-upload output through cli-logger Follows the repo convention: human-facing progress goes through src/utils/cli-logger.ts, not a migrate-private stderr logger. --- src/migrate/direct-upload.ts | 52 ++++++++++++++++++------------------ 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/src/migrate/direct-upload.ts b/src/migrate/direct-upload.ts index de6a11fb..6f88617e 100644 --- a/src/migrate/direct-upload.ts +++ b/src/migrate/direct-upload.ts @@ -37,7 +37,7 @@ import { } from './gc-window.js' import { formatDuration, Timer } from './metrics.js' import { type AddPiecesEvent, dataSetPieceId, fetchAddPiecesEvent, txLanded } from './pdp-verifier.js' -import { log } from './util.js' +import { log } from '../utils/cli-logger.js' export interface DirectUploadOptions { /** Initialized Synapse instance (auth, chain, and transport already resolved). */ @@ -171,7 +171,7 @@ export const defaultDirectUploadDeps: DirectUploadDeps = { } catch (err) { // A resumed run may find the CAR already evicted by a prior run. if ((err as NodeJS.ErrnoException).code !== 'ENOENT') { - log(`warn: failed to evict cached CAR ${path}: ${(err as Error).message}`) + log.message(`warn: failed to evict cached CAR ${path}: ${(err as Error).message}`) } } }, @@ -214,7 +214,7 @@ export async function runDirectUpload( const [primary, ...secondaries] = contexts if (primary == null) throw new Error('no primary storage context') - log( + log.message( `direct upload to ${contexts.length} provider(s): ` + contexts.map((c, i) => `${i === 0 ? 'primary' : 'secondary'} ${c.providerId} (${c.serviceURL})`).join(', ') ) @@ -232,7 +232,7 @@ export async function runDirectUpload( if (batch.length === 0) return flushCounts.set(ctx.providerId, (flushCounts.get(ctx.providerId) ?? 0) + 1) const cids = batch.map((b) => b.subPieceCid) - log(`flush [${reason}] provider ${ctx.providerId}: committing ${batch.length} piece(s)`) + log.message(`flush [${reason}] provider ${ctx.providerId}: committing ${batch.length} piece(s)`) // Durable breadcrumb before the attempt: a crash mid-commit must never be // auto-resolved into a blind re-add. db.markUploadsAddUnconfirmed(cids, ctx.providerId) @@ -254,14 +254,14 @@ export async function runDirectUpload( // commit; recording it here means later parks and resumes carry the // set id on the row instead of depending on context re-resolution. ctx.dataSetId = String(result.dataSetId) - log(`committed ${batch.length} piece(s) on provider ${ctx.providerId} (data set ${result.dataSetId})`) + log.message(`committed ${batch.length} piece(s) on provider ${ctx.providerId} (data set ${result.dataSetId})`) } catch (err) { const message = (err as Error).message ?? String(err) const gcCid = collectedCidFromError(message) if (gcCid == null) { // Not a GC rejection: leave the batch in add_unconfirmed for the // resume reconciliation: a blind retry could double-add on chain. - log(`error: commit failed on provider ${ctx.providerId} (batch left add_unconfirmed): ${message}`) + log.message(`error: commit failed on provider ${ctx.providerId} (batch left add_unconfirmed): ${message}`) return } // Curio rejected the batch because a parked piece is gone. The batch is @@ -270,12 +270,12 @@ export async function runDirectUpload( // Curio reports only the FIRST miss. const collected = batch.find((b) => b.subPieceCid === gcCid) if (collected == null) { - log(`warn: provider ${ctx.providerId} rejected unknown sub-piece ${gcCid}; re-verifying batch`) + log.message(`warn: provider ${ctx.providerId} rejected unknown sub-piece ${gcCid}; re-verifying batch`) } else { const age = deps.now() - Date.parse(collected.parkedAt) const lowered = lowerWindowOnGc(windowFor(ctx), age) db.lowerProviderWindow(ctx.providerId, lowered) - log( + log.message( `GC detected on provider ${ctx.providerId}: ${gcCid} collected after ${formatDuration(age)} parked; ` + `window lowered to ${formatDuration(lowered)}` ) @@ -286,7 +286,7 @@ export async function runDirectUpload( db.revertUploadsToParked([b.subPieceCid], ctx.providerId) } else { db.markUploadCollected(b.subPieceCid, ctx.providerId) - log(`collected: ${b.subPieceCid} on provider ${ctx.providerId} (will re-store)`) + log.message(`collected: ${b.subPieceCid} on provider ${ctx.providerId} (will re-store)`) } } } @@ -369,14 +369,14 @@ export async function runDirectUpload( } catch (err) { if (err instanceof CommPMismatchError) { db.markUploadFailed(next.subPieceCid, primary.providerId, 'primary', err.message) - log(`error: ${err.message}`) + log.message(`error: ${err.message}`) continue } throw err } storedBytes += stored.size db.recordUploadParked(next.subPieceCid, primary.providerId, 'primary', primary.dataSetId) - log( + log.message( `parked ${next.subPieceCid} (${formatFileSize(stored.size)}) on primary ${primary.providerId} ` + `in ${formatDuration(storeTimer.stop())}` ) @@ -411,7 +411,7 @@ export async function runDirectUpload( .map((u) => ({ ctx, u })) ) if (needsRetry.length === 0 && missingSecondaries.length === 0) break - log(`retrying ${needsRetry.length + missingSecondaries.length} piece(s) that did not land (attempt ${attempt + 1})`) + log.message(`retrying ${needsRetry.length + missingSecondaries.length} piece(s) that did not land (attempt ${attempt + 1})`) for (const { ctx, subPieceCid } of missingSecondaries) { await pullToSecondary(db, primary, ctx, subPieceCid) } @@ -422,7 +422,7 @@ export async function runDirectUpload( } const sub = db.subPieceByCid(u.subPieceCid) if (sub?.carPath == null) { - log(`error: collected ${u.subPieceCid} has no local CAR; cannot re-store`) + log.message(`error: collected ${u.subPieceCid} has no local CAR; cannot re-store`) continue } try { @@ -432,7 +432,7 @@ export async function runDirectUpload( } catch (err) { if (err instanceof CommPMismatchError) { db.markUploadFailed(sub.subPieceCid, ctx.providerId, u.role, err.message) - log(`error: ${err.message}`) + log.message(`error: ${err.message}`) continue } throw err @@ -463,7 +463,7 @@ export async function runDirectUpload( storedBytes, evictedCars: evictedPaths.size, } - log(`direct upload finished in ${formatDuration(runTimer.stop())}: ${formatFileSize(storedBytes)} stored`) + log.message(`direct upload finished in ${formatDuration(runTimer.stop())}: ${formatFileSize(storedBytes)} stored`) return summary } @@ -521,7 +521,7 @@ async function reconcileUnconfirmed( // row stays unresolved and the run exits incomplete. const dataSetId = u.dataSetId ?? ctx.dataSetId if (dataSetId == null) { - log( + log.message( `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and no known data set ` + `on provider ${ctx.providerId}; left add_unconfirmed for manual resolution` ) @@ -530,7 +530,7 @@ async function reconcileUnconfirmed( const pieceId = await deps.dataSetPieceId(synapse, dataSetId, u.subPieceCid) if (pieceId != null) { db.markUploadCommitted(u.subPieceCid, ctx.providerId, { dataSetId, pieceId, txHash: null }) - log(`resume: ${u.subPieceCid} found on chain in data set ${dataSetId} (piece ${pieceId}); marked committed`) + log.message(`resume: ${u.subPieceCid} found on chain in data set ${dataSetId} (piece ${pieceId}); marked committed`) continue } // Absent on chain. A transaction the provider broadcast just before @@ -541,7 +541,7 @@ async function reconcileUnconfirmed( // resolves them one way or the other. const ageMs = deps.now() - Date.parse(u.updatedAt) if (ageMs < UNCONFIRMED_REQUEUE_AFTER_MS) { - log( + log.message( `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and is absent from ` + `data set ${dataSetId}; too recent to rule out an in-flight transaction, left add_unconfirmed ` + `(re-run after ${formatDuration(UNCONFIRMED_REQUEUE_AFTER_MS - ageMs)})` @@ -568,13 +568,13 @@ async function reconcileUnconfirmed( pieceId: String(event.pieceIds[pieceIndex] ?? ''), txHash: u.txHash, }) - log( + log.message( `resume: ${u.subPieceCid} confirmed on chain via PiecesAdded (tx ${u.txHash}, ` + `data set ${dataSetId}); marked committed` ) continue } - log( + log.message( `resume: ${u.subPieceCid} has a LANDED addPieces tx ${u.txHash} on provider ${ctx.providerId} ` + `but its PiecesAdded event could not be verified; leaving add_unconfirmed: check the data set ` + `on the explorer before any manual retry (a blind re-add would duplicate the piece)` @@ -587,7 +587,7 @@ async function reconcileUnconfirmed( // realistic confirmation window re-enters the flow. const ageMs = deps.now() - Date.parse(u.updatedAt) if (ageMs < UNCONFIRMED_REQUEUE_AFTER_MS) { - log( + log.message( `resume: ${u.subPieceCid} has an unlanded addPieces tx ${u.txHash} on provider ${ctx.providerId}; ` + `too recent to rule out an in-flight transaction, left add_unconfirmed ` + `(re-run after ${formatDuration(UNCONFIRMED_REQUEUE_AFTER_MS - ageMs)})` @@ -597,10 +597,10 @@ async function reconcileUnconfirmed( } if (await ctx.hasPiece(CID.parse(u.subPieceCid))) { db.revertUploadsToParked([u.subPieceCid], ctx.providerId) - log(`resume: ${u.subPieceCid} still parked on provider ${ctx.providerId}; re-queued for commit`) + log.message(`resume: ${u.subPieceCid} still parked on provider ${ctx.providerId}; re-queued for commit`) } else { db.markUploadCollected(u.subPieceCid, ctx.providerId) - log(`resume: ${u.subPieceCid} gone from provider ${ctx.providerId}; will re-store`) + log.message(`resume: ${u.subPieceCid} gone from provider ${ctx.providerId}; will re-store`) } } } @@ -623,14 +623,14 @@ async function pullToSecondary( }) if (pulled.status === 'complete') { db.recordUploadParked(subPieceCid, secondary.providerId, 'secondary', secondary.dataSetId) - log(`parked ${subPieceCid} on secondary ${secondary.providerId} (pulled from primary)`) + log.message(`parked ${subPieceCid} on secondary ${secondary.providerId} (pulled from primary)`) } else { db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', 'secondary pull failed') - log(`warn: secondary ${secondary.providerId} failed to pull ${subPieceCid}`) + log.message(`warn: secondary ${secondary.providerId} failed to pull ${subPieceCid}`) } } catch (err) { db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', (err as Error).message) - log(`warn: secondary ${secondary.providerId} pull error for ${subPieceCid}: ${(err as Error).message}`) + log.message(`warn: secondary ${secondary.providerId} pull error for ${subPieceCid}: ${(err as Error).message}`) } }