diff --git a/reference-implementations/dacs-directory/src/catalog/reindexCore.ts b/reference-implementations/dacs-directory/src/catalog/reindexCore.ts index 37b8f7f..fcad4e3 100644 --- a/reference-implementations/dacs-directory/src/catalog/reindexCore.ts +++ b/reference-implementations/dacs-directory/src/catalog/reindexCore.ts @@ -4,8 +4,14 @@ * chain state and rewrite the catalog cache. Used by the CLI (npm run index) * and by POST /api/dacs/reindex (the UI's refresh button). */ +import { createHash } from "node:crypto"; import { indexRegistration, type ResolveIdentities } from "./indexer"; -import { boundedRevocationCandidates, readChainTip, scanChain } from "./scan"; +import { + boundedRevocationCandidates, + readChainTip, + scanChain, + scanConsensusAnchorBackfill, +} from "./scan"; import { chainResetRequired, chainResetThreshold } from "./chainContinuity"; import { crawlDomains } from "./wellknown"; import { upsertCounterpartyEvidenceSeller } from "./counterpartyEvidence"; @@ -24,6 +30,8 @@ import { recordArtifact, pruneFailureHistory, recordArtifactFailure, + loadUnanchoredBundleTargets, + recordConsensusAnchors, } from "./store"; import type { Registration } from "./types"; import type { ResolveRecipe } from "./identityVerification"; @@ -66,7 +74,7 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise 0 ? configuredMax : 100000; const configuredOverlap = Number(process.env.DACS_SCAN_REPLAY_DEPTH ?? 2); @@ -149,6 +159,47 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise) => createHash("sha256") + .update([...targets].sort(([a], [b]) => a.localeCompare(b)).map(([locator, hash]) => `${locator}:${hash}`).join("\n")) + .digest("hex"); + let unresolvedKey = targetKey(anchorTargets); + if (state.anchorBackfillTargetKey !== unresolvedKey) { + state.anchorBackfillCursor = undefined; + state.anchorBackfillComplete = false; + state.anchorBackfillTargetKey = unresolvedKey; + } + if (anchorTargets.size > 0 && !state.anchorBackfillComplete) { + try { + const configuredBackfillMax = Number(process.env.DACS_ANCHOR_BACKFILL_MAX_TXS ?? 500); + const configuredBackfillBudget = Number(process.env.DACS_ANCHOR_BACKFILL_BUDGET_MS ?? 10_000); + const backfill = await scanConsensusAnchorBackfill(anchorTargets, { + cursor: state.anchorBackfillCursor, + maxTxs: configuredBackfillMax, + budgetMs: configuredBackfillBudget, + }); + const updated = recordConsensusAnchors(backfill.observations); + state.anchorBackfillCursor = backfill.nextCursor; + state.anchorBackfillComplete = backfill.complete; + anchorTargets = loadUnanchoredBundleTargets(); + unresolvedKey = targetKey(anchorTargets); + state.anchorBackfillTargetKey = unresolvedKey; + if (anchorTargets.size === 0) state.anchorBackfillComplete = true; + log(`SR-2 anchor backfill: scanned ${backfill.txsScanned} tx(s), resolved ${updated}, ` + + `${anchorTargets.size} bundle(s) remain${backfill.complete ? " (history exhausted)" : ""}`); + } catch (error) { + log(`SR-2 anchor backfill deferred: ${error instanceof Error ? error.message : String(error)}`); + } + } else if (anchorTargets.size === 0) { + state.anchorBackfillCursor = undefined; + state.anchorBackfillComplete = true; + state.anchorBackfillTargetKey = unresolvedKey; + } const nextCursor = Math.max(state.lastSeenTxId, scan.highestTxId); if (nextCursor > state.lastSeenTxId) state.cursorAdvancedAt = Date.now(); // Seed upgraded state once so an already-frozen cursor becomes diagnosable @@ -156,7 +207,7 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise { @@ -41,7 +42,7 @@ export interface ScannedArtifacts { /** True only when the walk reached sinceTxId/genesis rather than maxTxs/error. */ complete: boolean; chainTip: number; - observations: Array<{ locator: string; kind: string; profile: string; owner?: string; observedAt: number; anchorTime?: number; data?: Record }>; + observations: Array<{ locator: string; kind: string; profile: string; owner?: string; contentHash?: string; observedAt: number; anchorTime?: number; data?: Record }>; failures: Array<{ locator: string; kind: string; code: string; message: string }>; scanError?: string; } @@ -159,11 +160,11 @@ export function isListingRevocationCandidate(value: unknown): boolean { } /** Unauthenticated nodeCall (plain fetch — no demosdk in the scan path). */ -async function nodeCall(message: string, data: Record): Promise { +async function nodeCall(message: string, data: Record, timeoutMs = 30_000): Promise { const res = await fetch(RPC + "/", { method: "POST", headers: { "content-type": "application/json" }, - signal: AbortSignal.timeout(30_000), + signal: AbortSignal.timeout(timeoutMs), body: JSON.stringify({ method: "nodeCall", params: [{ type: "nodeCall", message, sender: null, receiver: null, timestamp: null, data, extra: "" }], @@ -174,6 +175,151 @@ async function nodeCall(message: string, data: Record): Promise return json.response; } +interface StorageWriteCandidate { + locator: string; + contentHash: string; + blockNumber: number; + transactionHash: string; +} + +interface ConsensusAnchorObservation { + locator: string; + contentHash: string; + anchorTime: number; +} + +const objectValue = (value: unknown): Record | null => + value !== null && typeof value === "object" && !Array.isArray(value) + ? value as Record + : null; + +const parsedObject = (value: unknown): Record | null => { + if (typeof value !== "string") return objectValue(value); + try { return objectValue(JSON.parse(value)); } catch { return null; } +}; + +/** Exact target/content attribution for a whole-value StorageProgram write. */ +export function storageWriteCandidate(value: unknown): StorageWriteCandidate | null { + const tx = objectValue(value); + if (!tx || tx.status !== "confirmed" || tx.type !== "storageProgram") return null; + if (!Number.isSafeInteger(tx.blockNumber) || Number(tx.blockNumber) < 0) return null; + if (typeof tx.hash !== "string" || !/^[0-9a-fA-F]{64}$/.test(tx.hash)) return null; + if (typeof tx.to !== "string" || !/^stor-[0-9a-f]{40}$/.test(tx.to)) return null; + const envelope = parsedObject(tx.content); + if (!envelope || envelope.type !== "storageProgram" || envelope.to !== tx.to) return null; + if (!Array.isArray(envelope.data) || envelope.data.length !== 2 || envelope.data[0] !== "storageProgram") return null; + const write = objectValue(envelope.data[1]); + if (!write || (write.operation !== "CREATE_STORAGE_PROGRAM" && write.operation !== "WRITE_STORAGE")) return null; + if (write.storageAddress !== tx.to) return null; + const data = objectValue(write.data); + if (!data) return null; + return { + locator: tx.to, + contentHash: contentHash(data), + blockNumber: Number(tx.blockNumber), + transactionHash: tx.hash.toLowerCase(), + }; +} + +interface ConfirmedBlock { + number: number; + timestamp: number; + transactionHashes: Set; +} + +async function confirmedBlock( + blockNumber: number, + cache: Map>, + timeoutMs = 30_000, +): Promise { + let pending = cache.get(blockNumber); + if (!pending) { + pending = (async () => { + const block = objectValue(await nodeCall("getBlockByNumber", { blockNumber }, timeoutMs)); + const body = parsedObject(block?.content); + if (!block || block.status !== "confirmed" || block.number !== blockNumber || !body) return null; + if (!Number.isSafeInteger(body.timestamp) || Number(body.timestamp) < 0) return null; + if (!Array.isArray(body.ordered_transactions)) return null; + const timestamp = Number(body.timestamp) * 1_000; + if (!Number.isSafeInteger(timestamp)) return null; + return { + number: blockNumber, + timestamp, + transactionHashes: new Set(body.ordered_transactions + .filter((hash): hash is string => typeof hash === "string") + .map((hash) => hash.toLowerCase())), + }; + })(); + cache.set(blockNumber, pending); + } + return pending; +} + +async function resolveConsensusAnchors( + candidates: StorageWriteCandidate[], + targets: ReadonlyMap, + deadline?: number, +): Promise { + const blocks = new Map>(); + const resolved = new Map(); + for (const candidate of candidates) { + if (targets.get(candidate.locator) !== candidate.contentHash) continue; + if (deadline !== undefined && Date.now() >= deadline) throw new Error("anchor backfill wall-clock budget exhausted"); + const timeoutMs = deadline === undefined ? 30_000 : Math.max(1, deadline - Date.now()); + const block = await confirmedBlock(candidate.blockNumber, blocks, timeoutMs); + if (!block?.transactionHashes.has(candidate.transactionHash)) continue; + const key = `${candidate.locator}\n${candidate.contentHash}`; + const prior = resolved.get(key); + if (!prior || block.timestamp < prior.anchorTime) { + resolved.set(key, { locator: candidate.locator, contentHash: candidate.contentHash, anchorTime: block.timestamp }); + } + } + return [...resolved.values()]; +} + +export interface AnchorBackfillResult { + observations: ConsensusAnchorObservation[]; + txsScanned: number; + nextCursor?: number; + complete: boolean; +} + +/** Bounded, resumable descending history scan for current bundle content. */ +export async function scanConsensusAnchorBackfill( + targets: ReadonlyMap, + opts: { cursor?: number; maxTxs?: number; budgetMs?: number } = {}, +): Promise { + if (targets.size === 0) return { observations: [], txsScanned: 0, complete: true }; + const maxTxs = Math.max(1, Math.min(5_000, nonNegativeInt(opts.maxTxs, 500))); + const budgetMs = Math.max(1_000, Math.min(60_000, nonNegativeInt(opts.budgetMs, 10_000))); + const deadline = Date.now() + budgetMs; + let cursor: number | "latest" = opts.cursor ?? "latest"; + let nextCursor: number | undefined = opts.cursor; + let scanned = 0; + let complete = false; + const candidates: StorageWriteCandidate[] = []; + while (scanned < maxTxs && Date.now() < deadline) { + const limit = Math.min(100, maxTxs - scanned); + const remaining = Math.max(1_000, deadline - Date.now()); + const page = ((await nodeCall("getTransactions", { start: cursor, limit }, remaining)) ?? []) as unknown[]; + if (page.length === 0) { complete = true; break; } + const ids = page.map((tx) => objectValue(tx)?.id) + .filter((id): id is number => Number.isSafeInteger(id) && Number(id) >= 0); + for (const tx of page) { + const candidate = storageWriteCandidate(tx); + if (candidate && targets.get(candidate.locator) === candidate.contentHash) candidates.push(candidate); + } + scanned += page.length; + if (ids.length === 0) throw new Error("anchor backfill page contained no valid transaction ids"); + const lowest = Math.min(...ids); + if (lowest <= 1) { complete = true; nextCursor = undefined; break; } + nextCursor = lowest - 1; + cursor = nextCursor; + } + const observations = await resolveConsensusAnchors(candidates, targets, deadline); + return { observations, txsScanned: scanned, nextCursor, complete }; +} + /** Read the node's current transaction tip without advancing scan state. */ export async function readChainTip(): Promise { const page = ((await nodeCall("getTransactions", { start: "latest", limit: 1 })) ?? []) as Array<{ id?: number }>; @@ -204,7 +350,7 @@ export async function scanChain( const since = nonNegativeInt(opts.sinceTxId, 0); const addresses = new Set(); for (const locator of opts.retryLocators ?? []) if (/^stor-[0-9a-f]{40}$/.test(locator)) addresses.add(locator); - const addressTimes = new Map(); + const writeCandidates: StorageWriteCandidate[] = []; let scanned = 0; let highestTxId = since; let complete = false; @@ -236,8 +382,9 @@ export async function scanChain( scanned += fresh.length; for (const tx of fresh) { const inTx = new Set(); collectNativeStorageAddresses(tx, inTx); - const timestamp = typeof (tx as { timestamp?: unknown }).timestamp === "number" ? (tx as { timestamp: number }).timestamp : undefined; - for (const address of inTx) { addresses.add(address); if (timestamp !== undefined && !addressTimes.has(address)) addressTimes.set(address, timestamp); } + for (const address of inTx) addresses.add(address); + const candidate = storageWriteCandidate(tx); + if (candidate) writeCandidates.push(candidate); } if (ids.length === 0) break; const lowest = Math.min(...ids); @@ -297,7 +444,20 @@ export async function scanChain( bundleOwners.set(name.slice("dacs5:bundle:".length), { address, owner: read.owner }); } observations.push({ locator: address, kind: artifactKind, profile: currentListing || currentBundle ? "dacs-v0.1" : "legacy-sdk-v0.1", owner: read.owner, - observedAt: Date.now(), anchorTime: addressTimes.get(address), data }); + contentHash: data ? contentHash(data) : undefined, observedAt: Date.now(), data }); + } + + const targets = new Map(observations + .filter((observation): observation is typeof observation & { contentHash: string } => + observation.kind === "bundle" && typeof observation.contentHash === "string") + .map((observation) => [observation.locator, observation.contentHash])); + try { + const anchors = await resolveConsensusAnchors(writeCandidates, targets); + const byLocator = new Map(anchors.map((anchor) => [anchor.locator, anchor.anchorTime])); + for (const observation of observations) observation.anchorTime = byLocator.get(observation.locator); + } catch { + // Consensus attribution is optional metadata. A block RPC failure must + // retain the honest finalisedAt fallback and will be retried by backfill. } // Attribute each discovered deal to its seller via the buyer-anchored agreement. diff --git a/reference-implementations/dacs-directory/src/catalog/store.ts b/reference-implementations/dacs-directory/src/catalog/store.ts index 66408f2..fe57705 100644 --- a/reference-implementations/dacs-directory/src/catalog/store.ts +++ b/reference-implementations/dacs-directory/src/catalog/store.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import { randomUUID } from "node:crypto"; import type { Catalog, Registration, ScanState } from "./types.js"; +import { deriveSellerReputation } from "./reputation.js"; export const DATA_DIR = process.env.DACS_DIRECTORY_DATA ?? join(process.cwd(), "data"); const DB_PATH = join(DATA_DIR, "directory.sqlite"); @@ -145,6 +146,31 @@ if (!(db.prepare("SELECT 1 FROM kv_state WHERE key='schema-version'").get())) db setJson("schema-version", 1); })(); +// Every persisted anchor time written before this migration came from the +// transaction's producer-controlled timestamp. Remove both the artifact rows +// and their already-materialised catalog derivatives atomically. Keeping the +// listings while recomputing seller reputation on the permitted finalisedAt +// fallback avoids serving stale SR-2 claims before the first v6 reindex; per- +// listing hints return only after that successful reindex. +if (getJson("sr2-anchor-schema-version", 0) < 2) db.transaction(() => { + db.prepare("UPDATE artifacts SET anchor_time = NULL").run(); + const catalog = getJson("catalog", { catalogVersion: "1", generatedAt: 0, sellers: [] }); + const windowEnd = catalog.generatedAt || Date.now(); + setJson("catalog", { + ...catalog, + sellers: catalog.sellers.map((seller) => { + const deals = seller.deals.map(({ anchorTimestamp: _unsafeAnchorTimestamp, ...deal }) => deal); + return { + ...seller, + deals, + listings: seller.listings.map(({ reputationHint: _unsafeReputationHint, ...listing }) => listing), + reputation: deriveSellerReputation(deals, 0, windowEnd), + }; + }), + }); + setJson("sr2-anchor-schema-version", 2); +})(); + const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); /** Cross-process/instance lease. SQLite serializes acquisition; expired leases recover automatically. */ @@ -224,7 +250,13 @@ const recordArtifactTransaction = db.transaction((observation: StoredArtifactObs VALUES (@locator,@kind,@profile,@owner,@contentHash,@observedAt,@anchorTime,@status,@dataJson) ON CONFLICT(locator) DO UPDATE SET kind=excluded.kind, profile=excluded.profile, owner=excluded.owner, content_hash=COALESCE(excluded.content_hash,artifacts.content_hash), observed_at=excluded.observed_at, - anchor_time=COALESCE(excluded.anchor_time,artifacts.anchor_time), status=excluded.status, + anchor_time=CASE + WHEN excluded.content_hash IS NOT NULL AND artifacts.content_hash IS NOT excluded.content_hash + THEN excluded.anchor_time + WHEN excluded.anchor_time IS NULL THEN artifacts.anchor_time + WHEN artifacts.anchor_time IS NULL THEN excluded.anchor_time + ELSE MIN(excluded.anchor_time,artifacts.anchor_time) + END, status=excluded.status, data_json=COALESCE(excluded.data_json,artifacts.data_json), error_code=NULL, error_message=NULL, retry_count=0, next_retry_at=NULL`) .run(observation); @@ -239,6 +271,34 @@ export function recordArtifact(observation: ArtifactObservation): void { status: observation.status ?? "observed", dataJson: observation.data ? JSON.stringify(observation.data) : null }); } +export interface ConsensusAnchorObservation { + locator: string; + contentHash: string; + anchorTime: number; +} + +/** + * Return only reputation-relevant artifacts that still need SR-2 time. The v6 + * history replay re-reads artifacts and supplies their current canonical hash + * before this query runs; rows without one cannot be safely attributed. + */ +export const loadUnanchoredBundleTargets = db.transaction((): Map => { + const rows = db.prepare(`SELECT locator,content_hash FROM artifacts + WHERE kind='bundle' AND anchor_time IS NULL AND status='observed' AND content_hash IS NOT NULL + ORDER BY locator`).all() as Array<{ locator: string; content_hash: string }>; + return new Map(rows.map((row) => [row.locator, row.content_hash])); +}); + +/** Keep the earliest consensus observation, but only for the exact current content. */ +export const recordConsensusAnchors = db.transaction((observations: ConsensusAnchorObservation[]): number => { + const update = db.prepare(`UPDATE artifacts SET anchor_time = CASE + WHEN anchor_time IS NULL THEN @anchorTime ELSE MIN(anchor_time,@anchorTime) END + WHERE locator=@locator AND content_hash=@contentHash`); + let changed = 0; + for (const observation of observations) changed += update.run(observation).changes; + return changed; +}); + const recordArtifactFailureTransaction = db.transaction( (locator: string, kind: string, code: string, message: string, maxRetries: number) => { const prior = db.prepare("SELECT retry_count,kind FROM artifacts WHERE locator = ?").get(locator) as { retry_count: number; kind: string } | undefined; diff --git a/reference-implementations/dacs-directory/src/catalog/types.ts b/reference-implementations/dacs-directory/src/catalog/types.ts index 2fff9cd..09f5990 100644 --- a/reference-implementations/dacs-directory/src/catalog/types.ts +++ b/reference-implementations/dacs-directory/src/catalog/types.ts @@ -232,6 +232,12 @@ export interface ScanState { lastChainTip?: number; /** Wall-clock time when lastSeenTxId most recently increased. */ cursorAdvancedAt?: number; + /** Descending transaction cursor for the bounded SR-2 consensus-time backfill. */ + anchorBackfillCursor?: number; + /** Hash of the unresolved bundle locator/content-hash set for this backfill cycle. */ + anchorBackfillTargetKey?: string; + /** True when the current unresolved target set has been searched to genesis. */ + anchorBackfillComplete?: boolean; /** owner + programName → observed native address (nonce-safe binding). */ programs?: Record; /** listing content hash → bounded, deterministic revocation candidates. */ diff --git a/reference-implementations/dacs-directory/test/sr2-anchor-migration.test.ts b/reference-implementations/dacs-directory/test/sr2-anchor-migration.test.ts new file mode 100644 index 0000000..bdfccf3 --- /dev/null +++ b/reference-implementations/dacs-directory/test/sr2-anchor-migration.test.ts @@ -0,0 +1,99 @@ +import assert from "node:assert/strict"; +import Database from "better-sqlite3"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +const dataDirectory = mkdtempSync(join(tmpdir(), "dacs-directory-sr2-migration-")); +const locator = `stor-${"7".repeat(40)}`; +const seeded = new Database(join(dataDirectory, "directory.sqlite")); +const seller = `did:demos:agent:${"8".repeat(64)}`; +seeded.exec(` + CREATE TABLE kv_state ( + key TEXT PRIMARY KEY, + value_json TEXT NOT NULL, + updated_at INTEGER NOT NULL + ); + INSERT INTO kv_state(key,value_json,updated_at) VALUES ('schema-version','1',0); + CREATE TABLE artifacts ( + locator TEXT PRIMARY KEY, + kind TEXT NOT NULL, + profile TEXT NOT NULL, + owner TEXT, + content_hash TEXT, + observed_at INTEGER NOT NULL, + anchor_time INTEGER, + status TEXT NOT NULL DEFAULT 'observed', + error_code TEXT, + error_message TEXT, + retry_count INTEGER NOT NULL DEFAULT 0, + next_retry_at INTEGER, + data_json TEXT + ); +`); +seeded.prepare(`INSERT INTO artifacts(locator,kind,profile,observed_at,anchor_time,status,data_json) + VALUES (?,?,?,?,?,'observed',?)`).run(locator, "bundle", "dacs-v0.1", 1, 123, JSON.stringify({ bundleVersion: "1" })); +seeded.prepare("INSERT INTO kv_state(key,value_json,updated_at) VALUES ('catalog',?,0)").run(JSON.stringify({ + catalogVersion: "1", + generatedAt: 1_000, + sellers: [{ + primaryClaim: seller, + displayName: "migration seller", + cci: [], + listings: [{ + listingId: "listing-1", + version: 1, + contentHash: "ab".repeat(32), + anchor: { kind: "storage-program", locator: `stor-${"9".repeat(40)}` }, + seller: { primaryClaim: seller, displayName: "migration seller" }, + offering: { title: "test", category: "test", tags: [] }, + pricing: {}, + status: "active", + catalogObservedAt: 1, + reputationHint: { + categoryScope: "test", completionRate: 1, bundleCount: 1, + windowStart: 0, windowEnd: 1_000, computedAt: 1_000, + }, + }], + deals: [{ + jobId: "job-1", + rail: "pay-dem", + buyerBundleRef: locator, + owners: { buyer: `did:demos:agent:${"6".repeat(64)}`, seller }, + signatureVerified: true, + refsVerified: true, + sellerOutcome: "completed", + finalisedAt: 100, + verifiedAt: 200, + reputationEligible: true, + anchorTimestamp: 123, + }], + reputation: { + completed: 1, + bundleCount: 1, + totalAgreements: 1, + completionRate: 1, + windowingBasis: "sr2-anchor-timestamp", + }, + registeredAt: 1, + lastIndexedAt: 1, + }], +})); +seeded.close(); + +// The migration runs while store.ts is imported, so the test database must be +// selected first. This also prevents tests from touching checkout-local data. +process.env.DACS_DIRECTORY_DATA = dataDirectory; +const store = await import("../src/catalog/store.js"); + +test.after(() => rmSync(dataDirectory, { recursive: true, force: true })); + +test("SR-2 migration removes every legacy producer-controlled anchor time once", () => { + assert.equal(store.artifactAnchorTime(locator), undefined); + const catalog = store.loadCatalog(); + assert.equal(catalog.sellers[0]?.deals[0]?.anchorTimestamp, undefined); + assert.equal(catalog.sellers[0]?.reputation.windowingBasis, "finalisedAt"); + assert.equal(catalog.sellers[0]?.reputation.bundleCount, 1); + assert.equal(catalog.sellers[0]?.listings[0]?.reputationHint, undefined); +}); diff --git a/reference-implementations/dacs-directory/test/sr2-anchor.test.ts b/reference-implementations/dacs-directory/test/sr2-anchor.test.ts new file mode 100644 index 0000000..915ff3f --- /dev/null +++ b/reference-implementations/dacs-directory/test/sr2-anchor.test.ts @@ -0,0 +1,133 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +const dataDirectory = mkdtempSync(join(tmpdir(), "dacs-directory-sr2-")); +const locator = `stor-${"a".repeat(40)}`; +const writeHash = "b".repeat(64); +const failedHash = "c".repeat(64); +const unrelatedHash = "d".repeat(64); +const bundle = { + bundleVersion: "1", + jobId: "job-sr2-test", + parties: [], + anchoredByRole: "buyer", +}; +const writeContent = JSON.stringify({ + type: "storageProgram", + to: locator, + data: ["storageProgram", { + operation: "WRITE_STORAGE", + storageAddress: locator, + data: bundle, + }], +}); +const transactions = [ + { id: 10, status: "confirmed", type: "transfer", hash: "e".repeat(64), blockNumber: 31, to: "0x1", content: "{}" }, + // This is the only content-producing transaction and the only valid anchor. + { id: 8, status: "confirmed", type: "storageProgram", hash: writeHash, blockNumber: 30, + to: locator, timestamp: 9, content: writeContent }, + // Earlier producer timestamps and mere references must not win. + { id: 7, status: "failed", type: "storageProgram", hash: failedHash, blockNumber: 20, + to: locator, timestamp: 1, content: writeContent }, + { id: 6, status: "confirmed", type: "transfer", hash: unrelatedHash, blockNumber: 19, + to: "0x2", timestamp: 2, content: JSON.stringify({ memo: locator }) }, + { id: 1, status: "confirmed", type: "transfer", hash: "f".repeat(64), blockNumber: 1, to: "0x3", content: "{}" }, +]; + +let blockReads = 0; +const server = createServer((req, res) => { + if (req.method === "GET" && req.url === `/storage-program/${locator}`) { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ success: true, owner: `0x${"1".repeat(64)}`, programName: "dacs5:bundle:job-sr2-test", data: bundle })); + return; + } + let body = ""; + req.setEncoding("utf8"); + req.on("data", (chunk) => { body += chunk; }); + req.on("end", () => { + const call = JSON.parse(body).params?.[0]; + res.setHeader("content-type", "application/json"); + if (call?.message === "getTransactions") { + const start = call.data?.start; + const limit = call.data?.limit ?? 100; + const page = transactions.filter((tx) => start === "latest" || tx.id <= start).slice(0, limit); + res.end(JSON.stringify({ result: 200, response: page })); + return; + } + if (call?.message === "getBlockByNumber" && call.data?.blockNumber === 30) { + blockReads += 1; + res.end(JSON.stringify({ result: 200, response: { + id: 31, + number: 30, + status: "confirmed", + hash: "9".repeat(64), + content: { timestamp: 1_785_920_618, ordered_transactions: [writeHash] }, + } })); + return; + } + res.end(JSON.stringify({ result: 200, response: null })); + }); +}); +await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); +const address = server.address(); +if (!address || typeof address === "string") throw new Error("test server did not bind"); + +// Import isolation is security-significant: the store opens SQLite at import. +process.env.DACS_DIRECTORY_DATA = dataDirectory; +process.env.DEMOS_RPC = `http://127.0.0.1:${address.port}`; +process.env.DACS_SCAN_FINALITY_DEPTH = "0"; + +const { contentHash } = await import("@kynesyslabs/dacs/canonical"); +const { scanChain, scanConsensusAnchorBackfill } = await import("../src/catalog/scan.js"); +const store = await import("../src/catalog/store.js"); + +test.after(() => { + server.close(); + rmSync(dataDirectory, { recursive: true, force: true }); +}); + +test("SR-2 uses confirmed block time and ignores failed or unrelated references", async () => { + const result = await scanChain(null, { maxTxs: 100, sinceTxId: 0 }); + const observation = result.observations.find((item) => item.locator === locator); + + assert.equal(observation?.anchorTime, 1_785_920_618_000); + assert.equal(observation?.contentHash, contentHash(bundle)); + assert.equal(blockReads, 1, "only the exact confirmed write causes a block lookup"); +}); + +test("SR-2 historical scanning is bounded and resumes from its returned cursor", async () => { + const targets = new Map([[locator, contentHash(bundle)]]); + const first = await scanConsensusAnchorBackfill(targets, { maxTxs: 2, budgetMs: 5_000 }); + + assert.equal(first.complete, false); + assert.equal(first.nextCursor, 7); + assert.equal(first.txsScanned, 2); + assert.deepEqual(first.observations, [{ locator, contentHash: contentHash(bundle), anchorTime: 1_785_920_618_000 }]); + + const second = await scanConsensusAnchorBackfill(targets, { cursor: first.nextCursor, maxTxs: 2, budgetMs: 5_000 }); + assert.equal(second.nextCursor, 5); + assert.equal(second.observations.length, 0); +}); + +test("stored anchor time is hash-bound and retains the earliest consensus observation", () => { + const first = { value: "first" }; + const second = { value: "second" }; + const storedLocator = `stor-${"2".repeat(40)}`; + store.recordArtifact({ locator: storedLocator, kind: "bundle", profile: "dacs-v0.1", + contentHash: contentHash(first), observedAt: 1, anchorTime: 200, data: first }); + store.recordArtifact({ locator: storedLocator, kind: "bundle", profile: "dacs-v0.1", + contentHash: contentHash(first), observedAt: 2, anchorTime: 300, data: first }); + assert.equal(store.artifactAnchorTime(storedLocator), 200); + + store.recordArtifact({ locator: storedLocator, kind: "bundle", profile: "dacs-v0.1", + contentHash: contentHash(second), observedAt: 3, data: second }); + assert.equal(store.artifactAnchorTime(storedLocator), undefined, "an overwrite cannot inherit the old content's time"); + + assert.equal(store.recordConsensusAnchors([{ locator: storedLocator, contentHash: contentHash(first), anchorTime: 100 }]), 0); + assert.equal(store.recordConsensusAnchors([{ locator: storedLocator, contentHash: contentHash(second), anchorTime: 400 }]), 1); + assert.equal(store.artifactAnchorTime(storedLocator), 400); +});