diff --git a/reference-implementations/dacs-directory/app/api/dacs/artifact/route.ts b/reference-implementations/dacs-directory/app/api/dacs/artifact/route.ts index e587b17..55b6aa8 100644 --- a/reference-implementations/dacs-directory/app/api/dacs/artifact/route.ts +++ b/reference-implementations/dacs-directory/app/api/dacs/artifact/route.ts @@ -8,8 +8,8 @@ * claim that those exact bytes occupy the requested native address. */ import { NextRequest, NextResponse } from "next/server"; -import { deriveAnchorAddress, readAnchor } from "@/src/catalog/chain"; -import { findProgramAddress } from "@/src/catalog/store"; +import { readAnchor } from "@/src/catalog/chain"; +import { resolveProgramAddress } from "@/src/catalog/store"; export async function GET(req: NextRequest) { const q = req.nextUrl.searchParams; @@ -17,7 +17,10 @@ export async function GET(req: NextRequest) { const owner = q.get("owner"); const name = q.get("name"); if (!ref && owner && name) { - ref = findProgramAddress(owner, name) ?? deriveAnchorAddress(owner, name); + ref = resolveProgramAddress(owner, name); + if (!ref) { + return NextResponse.json({ error: "more than one program uses this owner and name; the artifact is indeterminate" }, { status: 409 }); + } } if (!ref) return NextResponse.json({ error: "need ?ref= or ?owner=&name=" }, { status: 400 }); const value = await readAnchor(ref); diff --git a/reference-implementations/dacs-directory/app/api/dacs/deal-owners/route.ts b/reference-implementations/dacs-directory/app/api/dacs/deal-owners/route.ts index c90ff93..022a541 100644 --- a/reference-implementations/dacs-directory/app/api/dacs/deal-owners/route.ts +++ b/reference-implementations/dacs-directory/app/api/dacs/deal-owners/route.ts @@ -1,27 +1,64 @@ -/** GET /api/dacs/deal-owners?jobId= — catalog lookup for a deal's anchor owners +/** GET /api/dacs/deal-owners?jobId=[&bundleRef=] — catalog lookup for a deal's anchor owners * (used when a bundle copy doesn't name the buyer, e.g. seller-anchored copies). */ import { NextRequest, NextResponse } from "next/server"; -import { loadCatalog, loadScanState } from "@/src/catalog/store"; +import { canonicalDemosAgentClaim } from "@/src/catalog/claimRef"; +import { ownArray } from "@/src/catalog/scanState"; +import { artifactOwner, canonicalProgramOwner, loadCatalog, loadScanState } from "@/src/catalog/store"; +import type { RegisteredDeal } from "@/src/catalog/types"; +const party = (claim: string) => claim ? canonicalProgramOwner(claim) : ""; +const parties = (deal: RegisteredDeal) => `${party(deal.owners.buyer)}\n${party(deal.owners.seller)}`; +/** A deal without an attributed seller has no seller copy. */ +const sellerRef = (deal: RegisteredDeal) => deal.owners.seller ? deal.sellerBundleRef : undefined; + +/** + * A jobId is unique only per producer, so every catalog and scanned entry for it is + * considered. An entry counts only when each bundle ref it carries is bound to the party + * it is reported for: that party owns the storage program, or that party's verified + * BundleBinding for the role names the address. A catalog entry also has to sit in its + * own seller's record and either have both copies verified against these parties, or be + * a scanned entry as indexed: the same parties, each ref the scanned one or resolved + * through that party's binding. `bundleRef` then narrows the entries to the deal holding + * that copy. Owners are reported only when exactly one distinct, attributed deal remains. + */ export async function GET(req: NextRequest) { const jobId = req.nextUrl.searchParams.get("jobId")?.trim(); if (!jobId) return NextResponse.json({ error: "need ?jobId=" }, { status: 400 }); - for (const s of loadCatalog().sellers) { - const d = s.deals.find((x) => x.jobId === jobId); - if (d) { - return NextResponse.json({ - owners: d.owners, - buyerBundleRef: d.buyerBundleRef, - sellerBundleRef: d.sellerBundleRef ?? null, - }); - } + const bundleRef = req.nextUrl.searchParams.get("bundleRef")?.trim() || undefined; + const state = loadScanState(); + const bindings = ownArray(state.bundleBindings, jobId); + const signedFor = (ref: string, claim: string, role: "buyer" | "seller") => { + const holder = canonicalDemosAgentClaim(claim); + return holder !== null && bindings.some((binding) => + binding.role === role && binding.nativeAddress === ref && canonicalDemosAgentClaim(binding.signer) === holder); + }; + const boundTo = (ref: string, claim: string, role: "buyer" | "seller") => + canonicalDemosAgentClaim(claim) !== null && (artifactOwner(ref) === party(claim) || signedFor(ref, claim, role)); + const bound = (deal: RegisteredDeal) => boundTo(deal.buyerBundleRef, deal.owners.buyer, "buyer") && + (!sellerRef(deal) || boundTo(sellerRef(deal)!, deal.owners.seller, "seller")); + const entries = Object.values(state.deals).filter((deal) => deal.jobId === jobId); + const indexedFrom = (deal: RegisteredDeal) => entries.some((entry) => parties(entry) === parties(deal) && + (deal.buyerBundleRef === entry.buyerBundleRef || signedFor(deal.buyerBundleRef, deal.owners.buyer, "buyer")) && + (sellerRef(deal) === sellerRef(entry) || (!!sellerRef(deal) && signedFor(sellerRef(deal)!, deal.owners.seller, "seller")))); + const catalog = loadCatalog().sellers.flatMap((record) => record.deals.filter((deal) => deal.jobId === jobId && + canonicalDemosAgentClaim(deal.owners.seller) !== null && + canonicalDemosAgentClaim(deal.owners.seller) === canonicalDemosAgentClaim(record.primaryClaim) && + (deal.refsVerified || indexedFrom(deal)) && bound(deal))); + // A scanned entry already indexed into the catalog is represented by its catalog entry, + // whose bundle refs may have been resolved through bindings, whichever ref is requested. + const indexed = new Set(catalog.map(parties)); + const scanned = entries.filter((deal) => !indexed.has(parties(deal)) && bound(deal)); + const holds = (deal: RegisteredDeal) => !bundleRef || deal.buyerBundleRef === bundleRef || sellerRef(deal) === bundleRef; + const distinct = new Map(); + for (const deal of [...catalog, ...scanned].filter(holds)) { + distinct.set(JSON.stringify([party(deal.owners.buyer), party(deal.owners.seller), deal.buyerBundleRef, sellerRef(deal) ?? null]), deal); } - const scanned = loadScanState().deals[jobId]; - if (scanned) { + const [only] = distinct.values(); + if (distinct.size === 1 && only.owners.seller) { return NextResponse.json({ - owners: scanned.owners, - buyerBundleRef: scanned.buyerBundleRef, - sellerBundleRef: scanned.sellerBundleRef ?? null, + owners: only.owners, + buyerBundleRef: only.buyerBundleRef, + sellerBundleRef: sellerRef(only) ?? null, }); } return NextResponse.json({ owners: null, buyerBundleRef: null }); diff --git a/reference-implementations/dacs-directory/app/deal/[ref]/page.tsx b/reference-implementations/dacs-directory/app/deal/[ref]/page.tsx index 5e0cedc..ce6f7e3 100644 --- a/reference-implementations/dacs-directory/app/deal/[ref]/page.tsx +++ b/reference-implementations/dacs-directory/app/deal/[ref]/page.tsx @@ -4,8 +4,8 @@ import type { Metadata } from "next"; import CopyText from "@/src/components/CopyText"; import VerifyAttestation from "@/src/components/VerifyAttestation"; import VerifyDeal from "@/src/components/VerifyDeal"; -import { deriveAnchorAddress, readAnchor } from "@/src/catalog/chain"; -import { findProgramAddress } from "@/src/catalog/store"; +import { readAnchor } from "@/src/catalog/chain"; +import { resolveProgramAddress } from "@/src/catalog/store"; const EXPLORER = "https://explorer.demos.sh"; @@ -45,7 +45,7 @@ export default async function Deal({ const deliveryName = `dacsx:delivery:${jobId}`; const sellerClaim = seller ? decodeURIComponent(seller) : null; const attestationAddress = sellerClaim && jobId - ? findProgramAddress(sellerClaim, deliveryName) ?? deriveAnchorAddress(sellerClaim, deliveryName) + ? resolveProgramAddress(sellerClaim, deliveryName) : null; const attestation = attestationAddress ? await readAnchor(attestationAddress) : null; const att = attestation as { diff --git a/reference-implementations/dacs-directory/app/verify/page.tsx b/reference-implementations/dacs-directory/app/verify/page.tsx index 56cfbb4..2c6c3dd 100644 --- a/reference-implementations/dacs-directory/app/verify/page.tsx +++ b/reference-implementations/dacs-directory/app/verify/page.tsx @@ -40,7 +40,7 @@ export default function VerifyPage() { // 2. seller-anchored copy → the catalog knows the deal's owners. if (!buyer && jobId) { - const o = await fetch(`/api/dacs/deal-owners?jobId=${encodeURIComponent(jobId)}`).then((r) => r.json()); + const o = await fetch(`/api/dacs/deal-owners?jobId=${encodeURIComponent(jobId)}&bundleRef=${encodeURIComponent(address)}`).then((r) => r.json()); buyer = o?.owners?.buyer ?? null; } // 3. manual override as last resort. diff --git a/reference-implementations/dacs-directory/src/catalog/bundleBinding.ts b/reference-implementations/dacs-directory/src/catalog/bundleBinding.ts index 616471a..dce01f5 100644 --- a/reference-implementations/dacs-directory/src/catalog/bundleBinding.ts +++ b/reference-implementations/dacs-directory/src/catalog/bundleBinding.ts @@ -34,6 +34,11 @@ export function bundleBindingRoleKey(jobId: string, role: BundleBinding["role"]) return `${jobId}\n${role}`; } +/** Discovery is bounded per authenticated signer, so one signer's bindings never use another's budget (BB-6). */ +export function bundleBindingOverflowKey(jobId: string, role: BundleBinding["role"], signer: string): string { + return `${bundleBindingRoleKey(jobId, role)}\n${canonicalDemosAgentClaim(signer) ?? signer}`; +} + function strictSignatureBytes(value: unknown): Uint8Array | null { if (typeof value !== "string" || !/^[A-Za-z0-9_-]{86}$/.test(value)) return null; try { @@ -107,8 +112,9 @@ const bindingOrder = (left: BundleBinding, right: BundleBinding): number => contentHash(left).localeCompare(contentHash(right)); /** - * Deterministic total-work ceiling. Overflow is sticky in ScanState and must - * make that side indeterminate; truncation can never manufacture absence. + * Deterministic work ceiling per signer of each jobId and role. Overflow is + * sticky in ScanState and must make that signer's side indeterminate; + * truncation can never manufacture absence. */ export function boundedBundleBindings( bindings: Iterable, @@ -118,7 +124,7 @@ export function boundedBundleBindings( for (const binding of bindings) unique.set(contentHash(binding), binding); const byRole = new Map(); for (const binding of unique.values()) { - const key = bundleBindingRoleKey(binding.jobId, binding.role); + const key = bundleBindingOverflowKey(binding.jobId, binding.role, binding.signer); const values = byRole.get(key) ?? []; values.push(binding); byRole.set(key, values); diff --git a/reference-implementations/dacs-directory/src/catalog/bundlePolicy.ts b/reference-implementations/dacs-directory/src/catalog/bundlePolicy.ts index d528fc0..20fb467 100644 --- a/reference-implementations/dacs-directory/src/catalog/bundlePolicy.ts +++ b/reference-implementations/dacs-directory/src/catalog/bundlePolicy.ts @@ -98,15 +98,20 @@ export function bundleMatchesRegisteredDeal( sellers[0].primaryClaim === catalogSeller; } -/** Keep at most one verified reputation record per signed job and bundle ref. */ +/** + * Keep at most one verified reputation record per buyer and signed job, and per + * bundle ref. A jobId is unique only per producer, so two buyers' deals with + * the same jobId are distinct records. + */ export function dedupeVerifiedDeals(deals: DealRecord[]): DealRecord[] { const jobs = new Set(); const refs = new Set(); return deals.filter((deal) => { if (!deal.refsVerified) return true; + const job = `${demosSigningIdentity(deal.owners.buyer)}\n${deal.jobId}`; const ref = deal.buyerBundleRef.toLowerCase(); - if (jobs.has(deal.jobId) || refs.has(ref)) return false; - jobs.add(deal.jobId); + if (jobs.has(job) || refs.has(ref)) return false; + jobs.add(job); refs.add(ref); return true; }); diff --git a/reference-implementations/dacs-directory/src/catalog/chain.ts b/reference-implementations/dacs-directory/src/catalog/chain.ts index 056e7cf..181d851 100644 --- a/reference-implementations/dacs-directory/src/catalog/chain.ts +++ b/reference-implementations/dacs-directory/src/catalog/chain.ts @@ -122,7 +122,7 @@ export async function resolveOwnedAnchorByName( let candidates: ProgramCandidate[]; try { - const body = await response.json() as { result?: unknown; response?: unknown }; + const body = await storageResponseJson(response) as { result?: unknown; response?: unknown }; if (body.result !== 200 || !Array.isArray(body.response)) { return { status: "indeterminate", reason: "program-name lookup returned an invalid response" }; } @@ -154,7 +154,7 @@ export async function resolveOwnedAnchorByName( } let record: AnchorRecord | null = null; try { - const body = await read.json() as { + const body = await storageResponseJson(read) as { success?: boolean; data?: Record; owner?: string; @@ -162,7 +162,7 @@ export async function resolveOwnedAnchorByName( }; if ( body.success && body.data != null && typeof body.data === "object" && !Array.isArray(body.data) && - body.programName === programName + boundedJson(body.data) !== null && body.programName === programName ) { record = { data: body.data, owner: body.owner, programName: body.programName }; } diff --git a/reference-implementations/dacs-directory/src/catalog/indexer.ts b/reference-implementations/dacs-directory/src/catalog/indexer.ts index 7cf5a6b..4b3613d 100644 --- a/reference-implementations/dacs-directory/src/catalog/indexer.ts +++ b/reference-implementations/dacs-directory/src/catalog/indexer.ts @@ -24,9 +24,15 @@ import { verifyBundleCore } from "../../vendor/dacs-sdk/dist/agent/verifyBundleC // The SDK doesn't export sessionAnchorName from its public barrel // (dacs-sdk#14) — reach into the vendored build. import { sessionAnchorName } from "../../vendor/dacs-sdk/dist/agent/runSessionCore.js"; -import { deriveAnchorAddress, readAnchor, readAnchorRecord } from "./chain.js"; +import { readAnchor, readAnchorRecord } from "./chain.js"; import { gcrGetIdentities } from "./gcr.js"; -import { findValidListingRevocation, ownerClaim, verifyListing } from "./listingVerification.js"; +import { + findValidListingRevocation, + ownerClaim, + REVOCATION_VERIFICATIONS_PER_PASS, + verifyListing, + type RevocationCandidateProgress, +} from "./listingVerification.js"; import { canonicalDemosAgentClaim } from "./claimRef.js"; import { resolveDemosPrimaryClaimKey } from "./primaryClaimKey.js"; import { listingPresentation } from "./listingMetadata.js"; @@ -34,9 +40,11 @@ import { verifyOwnerSignature } from "./registrationSig.js"; import { artifactAnchorTime, clearListingRejection, - findProgramAddress, + nextRevocationCandidates, + resolveProgramAddress, loadScanState, recordListingRejection, + recordRevocationProgress, type ListingRejectionCode, throwIfInfrastructureError, } from "./store.js"; @@ -45,7 +53,7 @@ import { agreementPrice, buildCurrentEvidenceGraph, type EvidenceGraph } from ". import { agreementRail } from "./agreementMetadata.js"; import { currentBundleCopiesDiverge, reconcileCurrentCopies } from "./currentReconciliation.js"; import { - bundleBindingRoleKey, + bundleBindingOverflowKey, resolveBundleSide, verifyBundleBinding, } from "./bundleBinding.js"; @@ -139,7 +147,7 @@ export async function indexRegistration( const listingArtifacts = new Map }>(); let identityBundle: Record | undefined; const identityBundles: Record[] = []; - const revocations = loadScanState().revocations ?? {}; + const { verifiedRevocations } = loadScanState(); for (const anchor of reg.listingAnchors) { const anchored = await readAnchorRecord(anchor); if (!anchored) continue; @@ -168,16 +176,24 @@ export async function indexRegistration( : 1; const validity = scope.validity as { notAfter?: unknown } | undefined; if (typeof validity?.notAfter === "number" && validity.notAfter < now) continue; - const storedCandidates = revocations[verified.contentHash]; - const revocationAddresses = Array.isArray(storedCandidates) - ? storedCandidates - : storedCandidates ? [storedCandidates] : []; + // Verification work is bounded per pass: markers already verified come first (even when + // an older snapshot omitted them from the queue), then the next pending candidates, + // those anchored by the listing's own owner first. Only one batch is ever selected. + const limit = REVOCATION_VERIFICATIONS_PER_PASS; + const progress: RevocationCandidateProgress = { unread: [], rejected: [] }; const revocationBinding = await findValidListingRevocation( - revocationAddresses, + [ + ...(verifiedRevocations[verified.contentHash] ?? []).slice(0, limit), + ...nextRevocationCandidates(verified.contentHash, anchored.owner ?? null, limit), + ], verified, version, readAnchor, + undefined, + progress, + limit, ); + recordRevocationProgress(verified.contentHash, progress); const presentation = listingPresentation(scope); const signedSeller = scope.seller && typeof scope.seller === "object" && !Array.isArray(scope.seller) ? scope.seller as Record : null; @@ -223,15 +239,28 @@ export async function indexRegistration( const scanState = loadScanState(); const overflowBindings = new Set(scanState.bundleBindingOverflow ?? []); const relevantJobs = new Set((reg.deals ?? []).map((deal) => deal.jobId)); + // Only bindings signed by a deal's own buyer or seller are verified, so other signers' + // bindings for the same jobId add no work here (BB-6). + const holders = new Set((reg.deals ?? []).flatMap((deal) => [deal.owners.buyer, deal.owners.seller] + .map((claim) => `${deal.jobId}\n${canonicalDemosAgentClaim(claim)}`))); + const signedByHolder = (binding: BundleBinding) => typeof binding?.signer === "string" && + canonicalDemosAgentClaim(binding.signer) !== null && holders.has(`${binding.jobId}\n${canonicalDemosAgentClaim(binding.signer)}`); const rawBindings = [ ...[...relevantJobs].flatMap((jobId) => ownArray(scanState.bundleBindings, jobId)), ...(reg.bundleBindings ?? []).filter((binding) => relevantJobs.has(binding.jobId)), - ]; + ].filter(signedByHolder); const verifiedBindings = (await Promise.all(rawBindings.map((binding) => verifyBundleBinding(binding)))) .filter((binding): binding is BundleBinding => binding !== null); onOmitted?.("bindings", rawBindings.length - verifiedBindings.length); const indexDeal = async (deal: RegisteredDeal): Promise => { - const jobBindings = verifiedBindings.filter((binding) => binding.jobId === deal.jobId); + // Only bindings signed by this deal's role holders are discovery for it (BB-6). + const roleHolder = (role: string) => role === "buyer" ? deal.owners.buyer : role === "seller" ? deal.owners.seller : null; + const jobBindings = verifiedBindings.filter((binding) => { + const holder = roleHolder(binding.role); + return binding.jobId === deal.jobId && holder !== null && + canonicalDemosAgentClaim(binding.signer) !== null && + canonicalDemosAgentClaim(binding.signer) === canonicalDemosAgentClaim(holder); + }); const buyerInitial = await readAnchor(deal.buyerBundleRef); const resolveListing = async (ref: Record) => { const id = String(ref.listingId ?? ""); @@ -295,7 +324,7 @@ export async function indexRegistration( role: "buyer", expectedSigner: deal.owners.buyer, bindings: jobBindings, - overflow: overflowBindings.has(bundleBindingRoleKey(deal.jobId, "buyer")), + overflow: overflowBindings.has(bundleBindingOverflowKey(deal.jobId, "buyer", deal.owners.buyer)), inspect, }); const sellerResolution = await resolveBundleSide({ @@ -303,7 +332,7 @@ export async function indexRegistration( role: "seller", expectedSigner: deal.owners.seller, bindings: jobBindings, - overflow: overflowBindings.has(bundleBindingRoleKey(deal.jobId, "seller")), + overflow: overflowBindings.has(bundleBindingOverflowKey(deal.jobId, "seller", deal.owners.seller)), inspect, }); if (buyerResolution.disposition !== "present" || sellerResolution.disposition !== "present") { @@ -379,7 +408,8 @@ export async function indexRegistration( : kind === "dacs-4-evidence" ? sessionAnchorName.evidence(jobId) : kind === "dacs-2-verifyresult" ? sessionAnchorName.vet(jobId) : null; if (!name) return null; - const address = findProgramAddress(deal.owners.buyer, name) ?? deriveAnchorAddress(deal.owners.buyer, name); + const address = resolveProgramAddress(deal.owners.buyer, name); + if (!address) return null; const raw = await readAnchor(address); if (raw) resolvedArtifacts.push({ kind, raw }); return raw; diff --git a/reference-implementations/dacs-directory/src/catalog/listingVerification.ts b/reference-implementations/dacs-directory/src/catalog/listingVerification.ts index c7cff62..5863ac6 100644 --- a/reference-implementations/dacs-directory/src/catalog/listingVerification.ts +++ b/reference-implementations/dacs-directory/src/catalog/listingVerification.ts @@ -248,9 +248,20 @@ export function revocationLogicalAddress( return `dacs1-revoked:${encodedClaim}:${listingId}:v${listingVersion}`; } +/** Per-listing, per-pass bound on candidate reads and RB-4 verifications. */ +export const REVOCATION_VERIFICATIONS_PER_PASS = 16; + +/** Candidates examined by one call: unreadable ones are retried later, rejected ones were read and failed RB-4. */ +export interface RevocationCandidateProgress { + unread: string[]; + rejected: string[]; +} + /** - * Fully verify discovered candidates and return the RB-2 binding this catalog - * can publish. A bogus candidate never shadows a later valid marker. + * Fully verify discovered candidates, in the given order and at most `limit` + * of them, and return the RB-2 binding this catalog can publish. A bogus + * candidate never shadows a later valid marker; unexamined candidates are + * left for a later call. */ export async function findValidListingRevocation( candidateRefs: string[], @@ -258,12 +269,19 @@ export async function findValidListingRevocation( expectedVersion: number, readCandidate: (ref: string) => Promise | null>, resolveKey: ResolvePrimaryClaimKey = resolveDemosPrimaryClaimKey, + progress: RevocationCandidateProgress = { unread: [], rejected: [] }, + limit = REVOCATION_VERIFICATIONS_PER_PASS, ): Promise { - for (const ref of candidateRefs) { + for (const ref of [...new Set(candidateRefs)].slice(0, limit)) { const candidate = await readCandidate(ref); - if (!candidate || !(await verifyListingRevocation( - candidate, listing, expectedVersion, resolveKey, - ))) continue; + if (!candidate) { + progress.unread.push(ref); + continue; + } + if (!(await verifyListingRevocation(candidate, listing, expectedVersion, resolveKey))) { + progress.rejected.push(ref); + continue; + } const scope = stripSignature(candidate); return { sellerPrimaryClaim: listing.sellerClaim, diff --git a/reference-implementations/dacs-directory/src/catalog/registration.ts b/reference-implementations/dacs-directory/src/catalog/registration.ts index 6c46ca0..8af7f6f 100644 --- a/reference-implementations/dacs-directory/src/catalog/registration.ts +++ b/reference-implementations/dacs-directory/src/catalog/registration.ts @@ -3,6 +3,7 @@ import type { BundleBinding, RegisteredDeal, Registration } from "./types.js"; import { isCanonicalDemosAgentClaim } from "./claimRef.js"; const ANCHOR = /^stor-[0-9a-f]{40}$/; +export const MAX_REGISTRATION_BUNDLE_BINDINGS = 256; const stringField = (v: unknown, max: number): v is string => typeof v === "string" && v.trim().length > 0 && v.length <= max; @@ -63,7 +64,7 @@ export function parseRegistration(v: unknown): RegistrationParseResult { } if ( b.bundleBindings !== undefined && - (!Array.isArray(b.bundleBindings) || b.bundleBindings.length > 256 || + (!Array.isArray(b.bundleBindings) || b.bundleBindings.length > MAX_REGISTRATION_BUNDLE_BINDINGS || b.bundleBindings.some((binding) => !binding || typeof binding !== "object" || Array.isArray(binding) || boundedJson(binding, 16_384) === null)) ) { diff --git a/reference-implementations/dacs-directory/src/catalog/reindexCore.ts b/reference-implementations/dacs-directory/src/catalog/reindexCore.ts index 583e6e2..d687442 100644 --- a/reference-implementations/dacs-directory/src/catalog/reindexCore.ts +++ b/reference-implementations/dacs-directory/src/catalog/reindexCore.ts @@ -4,13 +4,12 @@ * 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 { normalizeScanState, ownArray } from "./scanState.js"; -import { parseRegistration } from "./registration.js"; +import { discoveredDealKey, normalizeScanState, ownArray, SCAN_STATE_SCHEMA_VERSION } from "./scanState.js"; +import { MAX_REGISTRATION_BUNDLE_BINDINGS, parseRegistration } from "./registration.js"; import { createHash } from "node:crypto"; import { indexRegistration, type ResolveIdentities } from "./indexer"; import { boundedBundleBindings, verifyBundleBinding } from "./bundleBinding"; import { - boundedRevocationCandidates, readChainTip, scanChain, scanConsensusAnchorBackfill, @@ -25,6 +24,7 @@ import { loadFixtureSeeds, loadRegistrations, loadScanState, + admitRevocationCandidates, saveCatalog, saveScanState, beginScanRun, @@ -38,7 +38,7 @@ import { recordConsensusAnchors, throwIfInfrastructureError, } from "./store"; -import type { Registration } from "./types"; +import type { BundleBinding, Registration } from "./types"; import type { ResolveRecipe } from "./identityVerification"; export interface ReindexSummary { @@ -62,11 +62,20 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise verifyBundleBinding(binding)))).filter((binding) => binding !== null) - : undefined; - if (bindings) omitted.omittedBindings += raw.bundleBindings!.length - bindings.length; - const parsed = parseRegistration(bindings ? { ...raw, bundleBindings: bindings } : raw); + // The registration and its binding count are checked before any binding is verified; + // admitted bindings are then verified one at a time. + const rawBindings: unknown = raw?.bundleBindings; + const admitted = Array.isArray(rawBindings) && rawBindings.length <= MAX_REGISTRATION_BUNDLE_BINDINGS ? rawBindings : undefined; + let parsed = parseRegistration(admitted ? { ...raw, bundleBindings: [] } : raw); + if (parsed.ok && admitted) { + const bindings: BundleBinding[] = []; + for (const binding of admitted) { + const verified = await verifyBundleBinding(binding); + if (verified) bindings.push(verified); + } + omitted.omittedBindings += admitted.length - bindings.length; + parsed = parseRegistration({ ...raw, bundleBindings: bindings }); + } if (!parsed.ok) { omitted.omittedSellers++; const claim = typeof raw?.primaryClaim === "string" ? raw.primaryClaim.slice(0, 24) : "invalid claim"; @@ -97,7 +106,7 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise 0 ? configuredMax : 100000; const configuredOverlap = Number(process.env.DACS_SCAN_REPLAY_DEPTH ?? 2); @@ -151,23 +163,29 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise [hash, new Set(addresses)]), - ); const runId = beginScanRun(sinceTxId); let scan: Awaited> | undefined; try { scan = await scanChain(null, { maxTxs, sinceTxId, retryLocators: loadRetryableArtifacts(), - knownPrograms: new Map(Object.entries(state.programs)), verifiedRevocations }); + knownPrograms: new Map(Object.entries(state.programs)) }); if (!scan.complete) { throw new Error( scan.scanError ? `chain scan failed before reaching its cursor: ${scan.scanError}` : `chain scan hit DACS_SCAN_MAX_TXS=${maxTxs} before reaching its cursor; increase the limit so the catalog cannot skip history`, ); } - if (needsHistoryReplay) state.deals = Object.create(null); + // Before v9, cached deals could carry stale attribution, so they are rebuilt; later + // entries are kept and the replay adds or refreshes each owner's entry. + if (needsHistoryReplay && (state.schemaVersion ?? 0) < 9) state.deals = Object.create(null); for (const [addr, owner] of scan.listings) state.listings[addr] = owner; - for (const [jobId, deal] of scan.deals) state.deals[jobId] = deal; + for (const [key, deal] of scan.deals) { + // Seller copies are matched within one scan window; keep the stored one when this window has none. + const stored = Object.hasOwn(state.deals, key) ? state.deals[key] : undefined; + if (!deal.sellerBundleRef && stored?.sellerBundleRef && stored.owners.seller === deal.owners.seller) { + deal.sellerBundleRef = stored.sellerBundleRef; + } + state.deals[key] = deal; + } state.programs ??= Object.create(null); for (const [key, address] of scan.programs) state.programs[key] = address; // A later duplicate invalidates prior automatic attribution too. @@ -178,22 +196,15 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise [observation.locator, observation])); for (const [hash, addresses] of scan.revocations) { - const priorCandidates = state.revocations[hash]; - const prior = Array.isArray(priorCandidates) - ? priorCandidates - : priorCandidates ? [priorCandidates] : []; - const merged = boundedRevocationCandidates( - [...addresses, ...prior], - verifiedRevocations.get(hash), - ); - state.revocations[hash] = merged.candidates; - revocationCandidatesTruncated += merged.truncated; + admitRevocationCandidates(hash, addresses.map((locator) => + ({ locator, owner: observed.get(locator)?.owner, contentHash: observed.get(locator)?.contentHash }))); } for (const [jobId, bindings] of scan.bundleBindings) { const bounded = boundedBundleBindings([...ownArray(state.bundleBindings, jobId), ...bindings]); @@ -207,14 +218,6 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise 0) { - log(`revocation candidates: truncated ${revocationCandidatesTruncated} unverified locator(s) at the per-listing bound`); - } const didOf = (addr: string) => `did:demos:agent:${addr.replace(/^0x/, "")}`; const known = new Set(regs.map((r) => r.primaryClaim)); @@ -308,7 +308,8 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise d.jobId === deal.jobId)) reg.deals.push(deal); + const key = discoveredDealKey(deal.owners.buyer, deal.jobId); + if (!reg.deals.some((d) => discoveredDealKey(d.owners.buyer, d.jobId) === key)) reg.deals.push(deal); const bindings = ownArray(state.bundleBindings, deal.jobId); if (bindings.length > 0) { reg.bundleBindings ??= []; @@ -401,12 +402,6 @@ export async function reindexAll(opts: ReindexOptions = {}): Promise { export interface ScannedArtifacts { /** listing anchor address → owner address */ listings: Map; - /** jobId → discovered deal (buyer-anchored bundle + owners) */ + /** buyer owner + jobId (discoveredDealKey) → discovered deal (buyer-anchored bundle + owners) */ deals: Map; /** owner + programName → observed native address. */ programs: Map; - /** listing content hash → bounded, deterministic revocation candidates. */ + /** listing content hash → every revocation candidate seen in this window, oldest first. */ revocations: Map; - /** Candidate locators discarded by the per-listing resource bound. */ - revocationCandidatesTruncated: number; /** jobId → BB-4-verified BundleBindings discovered in this scan window. */ bundleBindings: Map; /** jobId + role keys whose deterministic total-work cap was exhausted. */ @@ -56,7 +55,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; contentHash?: 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; rejected?: boolean }>; failures: Array<{ locator: string; kind: string; code: string; message: string }>; scanError?: string; } @@ -123,13 +122,17 @@ export async function readStorage(address: string, attempts = 3): Promise MAX_STORAGE_NAME_LENGTH || body.owner.length > MAX_STORAGE_OWNER_LENGTH || + (body.data != null && !objectValue(body.data) && /^dacs[0-9]+[:-]/.test(body.programName))) { + return { success: false, failureCode: "ARTIFACT_REJECTED" }; + } + return { ...body, data: objectValue(body.data) ?? undefined }; + } lastFailure = failure; if (attempt < boundedAttempts) await new Promise((resolve) => setTimeout(resolve, 100 * 2 ** (attempt - 1))); } @@ -164,40 +167,15 @@ export function collectNativeStorageAddresses(value: unknown, out: Set, const didOf = (address: string): string => `did:demos:agent:${address.replace(/^0x/, "")}`; -export function addRevocationCandidate( - revocations: Map, - listingHash: string, - address: string, - verifiedAddresses: ReadonlySet = new Set(), -): number { - const candidates = revocations.get(listingHash) ?? []; - const merged = boundedRevocationCandidates([...candidates, address], verifiedAddresses); - revocations.set(listingHash, merged.candidates); - return merged.truncated; -} - -export const MAX_REVOCATION_CANDIDATES_PER_LISTING = 16; - /** - * Keep discovery state deterministic and bounded while never evicting a marker - * that already passed RB-4 verification. Verified markers are signer-controlled - * rather than public-shape-controlled, so they form the explicit bound exception. + * Admit every candidate locator once. Discovery is never bounded by count: the + * indexer bounds verification work per pass and drops a locator only after it + * was read and rejected (REVOCATION_VERIFICATIONS_PER_PASS). */ -export function boundedRevocationCandidates( - addresses: Iterable, - verifiedAddresses: ReadonlySet = new Set(), - limit = MAX_REVOCATION_CANDIDATES_PER_LISTING, -): { candidates: string[]; truncated: number } { - // Include the persisted verified set even when an older scan-state snapshot - // omitted that locator from its candidate array. - const unique = [...new Set([...verifiedAddresses, ...addresses])]; - const verified = unique.filter((address) => verifiedAddresses.has(address)); - const unverified = unique.filter((address) => !verifiedAddresses.has(address)); - const candidates = [ - ...verified, - ...unverified.slice(0, Math.max(0, limit - verified.length)), - ]; - return { candidates, truncated: unique.length - candidates.length }; +export function addRevocationCandidate(revocations: Map>, listingHash: string, address: string): void { + const candidates = revocations.get(listingHash) ?? new Set(); + candidates.add(address); + revocations.set(listingHash, candidates); } /** @@ -446,7 +424,6 @@ export async function scanChain( sinceTxId?: number; retryLocators?: string[]; knownPrograms?: ReadonlyMap; - verifiedRevocations?: ReadonlyMap>; } = {}, ): Promise { // Incremental: walk latest → sinceTxId (exclusive) and stop. First run @@ -502,15 +479,14 @@ export async function scanChain( const listings = new Map(); const programs = new Map(opts.knownPrograms); - const revocations = new Map(); - let revocationCandidatesTruncated = 0; + const revocations = new Map>(); const bundleBindings = new Map(); const bundleBindingOverflow = new Set(); const observations: ScannedArtifacts["observations"] = []; const failures: ScannedArtifacts["failures"] = []; const namesByLocator = new Map(); let omittedBindings = 0; - const bundleOwners = new Map(); // jobId → buyer bundle + const bundleOwners = new Map(); // owner + jobId → buyer bundle const sellerCopies = new Map>(); // preserve competing candidates const addSellerCopy = (jobId: string, address: string, owner: string) => { const copies = sellerCopies.get(jobId) ?? []; @@ -540,7 +516,7 @@ export async function scanChain( if (dataHash === null) { // No canonical form, so no DACS signature can cover it: unclassified storage. // Only bounded rejection metadata is kept, never the payload itself. - observations.push({ locator: address, kind: "other", profile, owner: read.owner, observedAt: Date.now() }); + observations.push({ locator: address, kind: "other", profile, owner: read.owner, observedAt: Date.now(), rejected: true }); failures.push({ locator: address, kind: "other", code: "ARTIFACT_REJECTED", message: "storage artifact has no canonical JSON form" }); continue; } @@ -561,13 +537,7 @@ export async function scanChain( for (const key of bounded.overflowKeys) bundleBindingOverflow.add(key); } else if (isListingRevocationCandidate(data)) { artifactKind = "listing-revocation"; - const listingHash = String(data!.listingContentHash).toLowerCase(); - revocationCandidatesTruncated += addRevocationCandidate( - revocations, - listingHash, - address, - opts.verifiedRevocations?.get(listingHash), - ); + addRevocationCandidate(revocations, String(data!.listingContentHash).toLowerCase(), address); } else if (name.startsWith("dacs1:listing:") || name.startsWith("dacs1-") || currentListing) { artifactKind = "listing"; listings.set(address, read.owner); @@ -576,13 +546,14 @@ export async function scanChain( const jobId = data!.jobId as string; const role = data!.anchoredByRole; if (role === "seller") addSellerCopy(jobId, address, read.owner); - else bundleOwners.set(jobId, { address, owner: read.owner }); + else bundleOwners.set(discoveredDealKey(read.owner, jobId), { jobId, address, owner: read.owner }); } else if (name.startsWith("dacs5:bundle:seller:")) { artifactKind = "bundle"; addSellerCopy(name.slice("dacs5:bundle:seller:".length), address, read.owner); } else if (name.startsWith("dacs5:bundle:")) { artifactKind = "bundle"; - bundleOwners.set(name.slice("dacs5:bundle:".length), { address, owner: read.owner }); + const jobId = name.slice("dacs5:bundle:".length); + bundleOwners.set(discoveredDealKey(read.owner, jobId), { jobId, address, owner: read.owner }); } observations.push({ locator: address, kind: artifactKind, profile, owner: read.owner, contentHash: dataHash, observedAt: Date.now(), data }); @@ -603,7 +574,7 @@ export async function scanChain( // Attribute each discovered deal to its seller via the buyer-anchored agreement. const deals = new Map(); - for (const [jobId, bundle] of bundleOwners) { + for (const [dealKey, { jobId, ...bundle }] of bundleOwners) { const bundleName = namesByLocator.get(bundle.address); if (bundleName && programs.get(programBindingKey(bundle.owner, bundleName)) === null) continue; const agreementAddress = programs.get(programBindingKey(bundle.owner, `dacs3:agreement:${jobId}`)); @@ -616,15 +587,15 @@ export async function scanChain( const sellerFromAgreement = agreementData?.agreementVersion !== undefined || agreementData?.payeeBoundAgreementVersion !== undefined ? agreementParties.find((party) => party?.role === "seller" && typeof party.primaryClaim === "string")?.primaryClaim as string | undefined : typeof agreementData?.seller === "string" ? agreementData.seller : undefined; - const candidates = sellerCopies.get(jobId) ?? []; + // A seller copy is attached only to a deal whose seller is attributed, and only when that seller owns it. const sellerCopy = sellerFromAgreement - ? candidates.find((copy) => didOf(copy.owner) === sellerFromAgreement) - : candidates.length === 1 ? candidates[0] : undefined; + ? (sellerCopies.get(jobId) ?? []).find((copy) => didOf(copy.owner) === sellerFromAgreement) + : undefined; const seller = sellerFromAgreement; const rail = agreementRail(agreementData) ?? ((agreementData?.price as { rail?: string } | undefined)?.rail) ?? (((agreementData?.terms as Record | undefined)?.price as { rail?: string } | undefined)?.rail) ?? "unknown"; - deals.set(jobId, { + deals.set(dealKey, { jobId, rail, buyerBundleRef: bundle.address, @@ -634,7 +605,9 @@ export async function scanChain( }); } - return { listings, deals, programs, revocations, revocationCandidatesTruncated, + // Transactions are walked newest first; candidates queue in chain order. + return { listings, deals, programs, + revocations: new Map([...revocations].map(([hash, candidates]) => [hash, [...candidates].reverse()])), bundleBindings, bundleBindingOverflow, omittedBindings, txsScanned: scanned, highestTxId, complete, chainTip, observations, failures, scanError }; } diff --git a/reference-implementations/dacs-directory/src/catalog/scanState.ts b/reference-implementations/dacs-directory/src/catalog/scanState.ts index 0501717..24884c3 100644 --- a/reference-implementations/dacs-directory/src/catalog/scanState.ts +++ b/reference-implementations/dacs-directory/src/catalog/scanState.ts @@ -15,21 +15,47 @@ const dictionary = (value: unknown, accept: (value: unknown) => boolean): Rec const strings = (value: unknown): value is string[] => Array.isArray(value) && value.every((v) => typeof v === "string"); const string = (value: unknown): boolean => typeof value === "string"; +/** Demos owners appear as both 0x addresses and did:demos:agent claims. */ +export const canonicalProgramOwner = (owner: string): string => { + const hex = owner.match(/([0-9a-fA-F]{64})$/)?.[1]; + return hex ? `0x${hex.toLowerCase()}` : owner.toLowerCase(); +}; + +/** + * Discovered deals are namespaced by the storage owner of the bundle program + * and the jobId, because jobId uniqueness is a producer obligation a consumer + * cannot rely on across unrelated writers. The key is an indexing namespace + * only: buyer attribution still comes from the validated agreement and its + * role evidence (DACS-5 §10.4.2, DACS-3 CA-7), never from storage ownership. + */ +export const discoveredDealKey = (buyer: string, jobId: string): string => `${canonicalProgramOwner(buyer)}\n${jobId}`; + +/** + * v10 replays the full deal history so entries that jobId-keyed state could + * not hold are rebuilt under their owner key. A pass records the version only + * after its complete replay is saved; an interrupted replay simply runs again. + */ +export const SCAN_STATE_SCHEMA_VERSION = 10; + /** JSON/SQLite restore prototypes: rebuild every attacker-keyed dictionary on ingress. */ export type NormalizedScanState = ScanState & Required>; export function normalizeScanState(state: ScanState): NormalizedScanState { + const deals = dictionary(state.deals, (v) => { + if (!v || typeof v !== "object") return false; + const deal = v as ScanState["deals"][string]; + return typeof deal.jobId === "string" && typeof deal.rail === "string" && + typeof deal.buyerBundleRef === "string" && !!deal.owners && + typeof deal.owners.buyer === "string" && typeof deal.owners.seller === "string" && boundedJson(v) !== null; + }); return { ...state, listings: dictionary(state.listings, string), - deals: dictionary(state.deals, (v) => { - if (!v || typeof v !== "object") return false; - const deal = v as ScanState["deals"][string]; - return typeof deal.jobId === "string" && typeof deal.rail === "string" && - typeof deal.buyerBundleRef === "string" && !!deal.owners && - typeof deal.owners.buyer === "string" && typeof deal.owners.seller === "string" && boundedJson(v) !== null; - }), + // Entries are rekeyed from their own buyer and jobId, which also migrates jobId-keyed state in place. + deals: dictionary(Object.fromEntries(Object.values(deals) + .map((deal) => [discoveredDealKey(deal.owners.buyer, deal.jobId), deal])), () => true), programs: dictionary(state.programs, (v) => v === null || typeof v === "string"), + // Legacy carriage only: the candidate queue lives in SQLite (store.ts). revocations: dictionary(state.revocations, (v) => typeof v === "string" || strings(v)), verifiedRevocations: dictionary(state.verifiedRevocations, strings), bundleBindings: dictionary(state.bundleBindings, (v) => Array.isArray(v)), diff --git a/reference-implementations/dacs-directory/src/catalog/store.ts b/reference-implementations/dacs-directory/src/catalog/store.ts index a07f80c..ff1467d 100644 --- a/reference-implementations/dacs-directory/src/catalog/store.ts +++ b/reference-implementations/dacs-directory/src/catalog/store.ts @@ -1,6 +1,7 @@ /** Transactional index repository backed by SQLite, with one-time JSON migration. */ import { boundedJson } from "./artifactLimits.js"; -import { normalizeScanState } from "./scanState.js"; +import { canonicalProgramOwner, normalizeScanState } from "./scanState.js"; +import { deriveAnchorAddress } from "./chain.js"; import Database from "better-sqlite3"; import { existsSync, mkdirSync, readFileSync } from "node:fs"; import { join } from "node:path"; @@ -123,6 +124,20 @@ db.exec(` ); CREATE INDEX IF NOT EXISTS listing_rejections_recent_idx ON listing_rejections(last_seen_at DESC, locator); + CREATE TABLE IF NOT EXISTS revocation_candidates ( + listing_hash TEXT NOT NULL, + locator TEXT NOT NULL, + owner TEXT, + content_hash TEXT, + seq INTEGER NOT NULL, + state TEXT NOT NULL DEFAULT 'pending', + PRIMARY KEY(listing_hash, locator) + ); + CREATE INDEX IF NOT EXISTS revocation_candidates_seq_idx ON revocation_candidates(seq); + CREATE INDEX IF NOT EXISTS revocation_candidates_pending_idx + ON revocation_candidates(listing_hash, seq) WHERE state = 'pending'; + CREATE INDEX IF NOT EXISTS revocation_candidates_owner_idx + ON revocation_candidates(listing_hash, owner, seq) WHERE state = 'pending'; `); // Add internal omission counters without changing the public diagnostic projection. @@ -133,6 +148,40 @@ db.transaction(() => { } }).immediate(); +/** Retry reads per pass. Transient failures are returned first. */ +export const RETRY_READS_PER_PASS = 100; +/** Rejected retries per pass: reserved whenever any are due, and never more than this. */ +export const REJECTED_RETRY_READS_PER_PASS = 20; +/** Ceiling on scheduled rejected retries. Overflow waits in arrival order and rotates in. */ +export const MAX_ACTIVE_REJECTED_RETRIES = 1_000; +const REJECTED_ACTIVE = "rejection_count > 0 AND deferred_at IS NULL AND status IN ('retry','dead-letter')"; + +// Rejected content and transient read failures are counted and scheduled as separate classes. +db.transaction(() => { + const columns = db.pragma("table_info(artifacts)") as Array<{ name: string }>; + if (!columns.some((entry) => entry.name === "rejection_count")) { + db.exec("ALTER TABLE artifacts ADD COLUMN rejection_count INTEGER NOT NULL DEFAULT 0"); + // Rows written with one shared counter: a rejected row's count was its rejections. + db.prepare(`UPDATE artifacts SET rejection_count=MAX(retry_count,1), retry_count=0, status='retry', + next_retry_at=COALESCE(next_retry_at,0) WHERE error_code='ARTIFACT_REJECTED'`).run(); + db.prepare("DELETE FROM dead_letters WHERE error_code='ARTIFACT_REJECTED'").run(); + } + if (!columns.some((entry) => entry.name === "deferred_at")) db.exec("ALTER TABLE artifacts ADD COLUMN deferred_at INTEGER"); + db.exec("CREATE INDEX IF NOT EXISTS artifacts_rejected_retry_idx ON artifacts(deferred_at, next_retry_at) WHERE rejection_count > 0"); + // At most MAX_ACTIVE_REJECTED_RETRIES stay scheduled, including rows migrated above: + // the earliest due stay, and the rest wait for a slot, returning in due order. + // Idempotent, so it also runs on every start. + const excess = (db.prepare(`SELECT COUNT(*) count FROM artifacts WHERE ${REJECTED_ACTIVE}`).get() as { count: number }).count - + MAX_ACTIVE_REJECTED_RETRIES; + if (excess > 0) { + const deferred = (db.prepare(`SELECT locator FROM artifacts WHERE ${REJECTED_ACTIVE} + ORDER BY COALESCE(next_retry_at,0) DESC, locator DESC LIMIT ?`).all(excess) as Array<{ locator: string }>).reverse(); + const defer = db.prepare("UPDATE artifacts SET deferred_at=?, next_retry_at=NULL WHERE locator=?"); + const start = Date.now() - deferred.length; + for (const [index, row] of deferred.entries()) defer.run(start + index, row.locator); + } +}).immediate(); + export function throwIfInfrastructureError(error: unknown): void { const code = error && typeof error === "object" && "code" in error ? String(error.code) : ""; if (code.startsWith("SQLITE_") || ["EIO", "ENOSPC", "EROFS", "EACCES"].includes(code)) throw error; @@ -186,6 +235,78 @@ if (getJson("sr2-anchor-schema-version", 0) < 2) db.transaction(() => { setJson("sr2-anchor-schema-version", 2); })(); +/** + * Revocation candidates, per listing content hash. Each locator is admitted + * once and stays pending until it is read and fails RB-4; a rejected locator + * is admitted again only when the chain shows it with different content. + * `seq` is arrival order; an unreadable candidate is given a new one, so it + * moves behind the others and is retried later. + */ +export interface RevocationCandidateObservation { locator: string; owner?: string | null; contentHash?: string | null } +const nextCandidateSeq = db.prepare("SELECT COALESCE(MAX(seq),0) + 1 seq FROM revocation_candidates"); +const upsertCandidate = db.prepare(`INSERT INTO revocation_candidates(listing_hash,locator,owner,content_hash,seq,state) + VALUES (@listingHash,@locator,@owner,@contentHash,@seq,'pending') + ON CONFLICT(listing_hash,locator) DO UPDATE SET + owner=COALESCE(excluded.owner,revocation_candidates.owner), + content_hash=COALESCE(excluded.content_hash,revocation_candidates.content_hash), + seq=CASE WHEN revocation_candidates.state='rejected' AND excluded.content_hash IS NOT NULL + AND excluded.content_hash IS NOT revocation_candidates.content_hash THEN excluded.seq ELSE revocation_candidates.seq END, + state=CASE WHEN revocation_candidates.state='rejected' AND excluded.content_hash IS NOT NULL + AND excluded.content_hash IS NOT revocation_candidates.content_hash THEN 'pending' ELSE revocation_candidates.state END`); +export const admitRevocationCandidates = db.transaction( + (listingHash: string, candidates: readonly RevocationCandidateObservation[]): void => { + let seq = (nextCandidateSeq.get() as { seq: number }).seq; + for (const candidate of candidates) { + upsertCandidate.run({ listingHash, locator: candidate.locator, seq: seq++, + owner: candidate.owner ? canonicalProgramOwner(candidate.owner) : null, contentHash: candidate.contentHash ?? null }); + } + }, +); +/** Candidate queues carried in scan-state JSON by earlier releases move into the table, in their stored order. */ +const ingestLegacyRevocations = (revocations: ScanState["revocations"]): void => { + const artifact = db.prepare("SELECT owner, content_hash FROM artifacts WHERE locator = ?"); + for (const [listingHash, stored] of Object.entries(revocations ?? {})) { + const locators = Array.isArray(stored) ? stored : stored ? [stored] : []; + admitRevocationCandidates(listingHash, locators.map((locator) => { + const row = artifact.get(locator) as { owner: string | null; content_hash: string | null } | undefined; + return { locator, owner: row?.owner, contentHash: row?.content_hash }; + })); + } +}; +db.transaction(() => { + const state = getJson("scan-state", null); + if (!state?.revocations || Object.keys(state.revocations).length === 0) return; + ingestLegacyRevocations(normalizeScanState(state).revocations); + setJson("scan-state", { ...state, revocations: {} }); +}).immediate(); + +const pendingCandidates = db.prepare(`SELECT locator FROM revocation_candidates + WHERE listing_hash = ? AND state = 'pending' ORDER BY seq LIMIT ?`); +const pendingOwnerCandidates = db.prepare(`SELECT locator FROM revocation_candidates + WHERE listing_hash = ? AND owner = ? AND state = 'pending' ORDER BY seq LIMIT ?`); +/** + * The next `limit` candidates to verify for a listing: those anchored by the + * listing's owner first, then the rest in arrival order. Two indexed queries + * of at most `limit` rows each, whatever the queue length. + */ +export function nextRevocationCandidates(listingHash: string, listingOwner: string | null, limit: number): string[] { + const rows = (statement: Database.Statement, ...params: unknown[]) => + (statement.all(...params) as Array<{ locator: string }>).map((row) => row.locator); + const owned = listingOwner ? rows(pendingOwnerCandidates, listingHash, canonicalProgramOwner(listingOwner), limit) : []; + return [...new Set([...owned, ...rows(pendingCandidates, listingHash, limit)])].slice(0, limit); +} +const rejectCandidate = db.prepare("UPDATE revocation_candidates SET state='rejected' WHERE listing_hash = ? AND locator = ?"); +const requeueCandidate = db.prepare(`UPDATE revocation_candidates SET seq = ? + WHERE listing_hash = ? AND locator = ? AND state = 'pending'`); +/** Apply one verification batch: rejected candidates leave the queue, unreadable ones move to its back. */ +export const recordRevocationProgress = db.transaction( + (listingHash: string, progress: { unread: readonly string[]; rejected: readonly string[] }): void => { + for (const locator of progress.rejected) rejectCandidate.run(listingHash, locator); + let seq = (nextCandidateSeq.get() as { seq: number }).seq; + for (const locator of progress.unread) requeueCandidate.run(seq++, listingHash, locator); + }, +); + const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); /** Cross-process/instance lease. SQLite serializes acquisition; expired leases recover automatically. */ @@ -220,7 +341,11 @@ export const saveCatalog = (catalog: Catalog): void => setJson("catalog", catalo export const loadRegistrations = (): Registration[] => getJson("registrations", []); export const saveRegistrations = (regs: Registration[]): void => setJson("registrations", regs); export const loadScanState = () => normalizeScanState(getJson("scan-state", { lastSeenTxId: 0, listings: {}, deals: {} })); -export const saveScanState = (state: ScanState): void => setJson("scan-state", normalizeScanState(state)); +export const saveScanState = db.transaction((state: ScanState): void => { + const normalized = normalizeScanState(state); + ingestLegacyRevocations(normalized.revocations); + setJson("scan-state", { ...normalized, revocations: {} }); +}); /** * Remove active cache rows that belong to a replaced chain. Registrations and * the append-only first-observation history survive so operator diagnostics do @@ -230,6 +355,7 @@ export const clearChainDerivedArtifacts = db.transaction((): void => { db.prepare("DELETE FROM listing_rejections").run(); db.prepare("DELETE FROM dead_letters").run(); db.prepare("DELETE FROM artifacts").run(); + db.prepare("DELETE FROM revocation_candidates").run(); }); export const loadDomains = (): string[] => getJson("domains", []); export const saveDomains = (domains: string[]): void => setJson("domains", [...new Set(domains)].sort()); @@ -242,22 +368,30 @@ export const loadFixtureSeeds = (): FixtureSeed[] => { export const saveFixtureSeeds = (seeds: FixtureSeed[]): void => setJson("fixtures", [...new Set(seeds)].filter((seed) => FIXTURE_SEEDS.has(seed)).sort()); -/** Demos owners appear as both 0x addresses and did:demos:agent claims. */ -export const canonicalProgramOwner = (owner: string): string => { - const hex = owner.match(/([0-9a-fA-F]{64})$/)?.[1]; - return hex ? `0x${hex.toLowerCase()}` : owner.toLowerCase(); -}; +export { canonicalProgramOwner }; export const programBindingKey = (owner: string, name: string): string => `${canonicalProgramOwner(owner)}\n${name}`; export const findProgramAddress = (owner: string, name: string): string | null => loadScanState().programs?.[programBindingKey(owner, name)] ?? null; - +/** + * Native address to read for an owner/name: the observed program, the legacy + * derived address when none was observed, or null when duplicate programs + * make the entry indeterminate. Null must never fall back to derivation. + */ +export function resolveProgramAddress(owner: string, name: string): string | null { + const programs = loadScanState().programs; + const key = programBindingKey(owner, name); + return Object.hasOwn(programs, key) ? programs[key] : deriveAnchorAddress(owner, name); +} export interface ArtifactObservation { locator: string; kind: string; profile: string; owner?: string; contentHash?: string; observedAt: number; anchorTime?: number; status?: string; data?: Record; + /** Set on the observation that accompanies a rejection; any other observation is a successful read. */ + rejected?: boolean; } -type StoredArtifactObservation = Omit & { +type StoredArtifactObservation = Omit & { contentHash: string | null; owner: string | null; anchorTime: number | null; status: string; dataJson: string | null; + rejected: number; }; const recordArtifactTransaction = db.transaction((observation: StoredArtifactObservation) => { @@ -272,9 +406,9 @@ const recordArtifactTransaction = db.transaction((observation: StoredArtifactObs WHEN artifacts.anchor_time IS NULL THEN excluded.anchor_time ELSE MIN(excluded.anchor_time,artifacts.anchor_time) END, status=excluded.status, - data_json=excluded.data_json, error_code=NULL, error_message=NULL, - retry_count=CASE WHEN excluded.content_hash IS NULL AND artifacts.error_code='ARTIFACT_REJECTED' - THEN artifacts.retry_count ELSE 0 END, next_retry_at=NULL`) + data_json=excluded.data_json, error_code=NULL, error_message=NULL, retry_count=0, + rejection_count=CASE WHEN @rejected THEN artifacts.rejection_count ELSE 0 END, + deferred_at=CASE WHEN @rejected THEN artifacts.deferred_at ELSE NULL END, next_retry_at=NULL`) .run(observation); // A readable observation is the recovery event for this locator. Keep the // active queue truthful and ensure a later transient failure starts at one. @@ -297,9 +431,10 @@ export function recordArtifact(observation: ArtifactObservation): void { contentHash = null; } } - recordArtifactTransaction({ ...observation, contentHash, + const { rejected, ...observed } = observation; + recordArtifactTransaction({ ...observed, contentHash, owner: observation.owner ?? null, anchorTime: contentHash ? observation.anchorTime ?? null : null, - status: observation.status ?? "observed", dataJson }); + status: observation.status ?? "observed", dataJson, rejected: rejected ? 1 : 0 }); } export interface ConsensusAnchorObservation { @@ -330,15 +465,39 @@ export const recordConsensusAnchors = db.transaction((observations: ConsensusAnc return changed; }); +const countActiveRejected = db.prepare(`SELECT COUNT(*) count FROM artifacts WHERE ${REJECTED_ACTIVE} AND locator <> ?`); +const longestWaitingRejected = db.prepare(`SELECT locator FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NOT NULL + AND locator <> ? ORDER BY deferred_at, locator LIMIT 1`); +const activateRejected = db.prepare("UPDATE artifacts SET deferred_at=NULL, next_retry_at=? WHERE locator=?"); + 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; - const attempts = (prior?.retry_count ?? 0) + 1; + const prior = db.prepare("SELECT retry_count,rejection_count,deferred_at,kind FROM artifacts WHERE locator = ?").get(locator) as + { retry_count: number; rejection_count: number; deferred_at: number | null; kind: string } | undefined; + const rejected = code === "ARTIFACT_REJECTED"; + const rejections = (prior?.rejection_count ?? 0) + (rejected ? 1 : 0); + // Content that was read ends a transient streak; transient failures have their own count. + const attempts = rejected ? 0 : (prior?.retry_count ?? 0) + 1; const failureKind = kind === "unknown" && prior?.kind && prior.kind !== "unknown" ? prior.kind : kind; - const periodic = code === "ARTIFACT_REJECTED"; + // Once its content was rejected, a locator stays periodic until a read succeeds. + const periodic = rejections > 0; const dead = !periodic && attempts >= maxRetries; const now = Date.now(); - const next = dead ? null : now + Math.min(60 * 60_000, 2 ** Math.min(attempts, 10) * 5_000); + let next: number | null = dead ? null : now + Math.min(60 * 60_000, 2 ** Math.min(rejected ? rejections : attempts, 10) * 5_000); + let deferredAt = periodic ? prior?.deferred_at ?? null : null; + if (rejected) { + const wasActive = (prior?.rejection_count ?? 0) > 0 && prior?.deferred_at == null; + const waiting = longestWaitingRejected.get(locator) as { locator: string } | undefined; + if (wasActive && waiting) { + // Rotation: a retried locator yields its slot to the longest-waiting one. + deferredAt = now; + activateRejected.run(now, waiting.locator); + } else if (!wasActive) { + const full = (countActiveRejected.get(locator) as { count: number }).count >= MAX_ACTIVE_REJECTED_RETRIES; + deferredAt = full ? deferredAt ?? now : null; + } + } + if (deferredAt !== null) next = null; const storedCode = code.slice(0, 100); const storedMessage = message.slice(0, 1000); db.prepare(`INSERT INTO artifact_failure_history(locator,first_seen_at,last_seen_at,observations) @@ -360,10 +519,12 @@ const recordArtifactFailureTransaction = db.transaction( } const history = db.prepare("SELECT first_seen_at FROM artifact_failure_history WHERE locator = ?") .get(locator) as { first_seen_at: number }; - db.prepare(`INSERT INTO artifacts(locator,kind,profile,observed_at,status,error_code,error_message,retry_count,next_retry_at) - VALUES (?,?,?,? ,?,?,?,?,?) ON CONFLICT(locator) DO UPDATE SET status=excluded.status,error_code=excluded.error_code, - error_message=excluded.error_message,retry_count=excluded.retry_count,next_retry_at=excluded.next_retry_at`) - .run(locator, failureKind, "unknown", now, dead ? "dead-letter" : "retry", storedCode, storedMessage, attempts, next); + db.prepare(`INSERT INTO artifacts(locator,kind,profile,observed_at,status,error_code,error_message,retry_count,next_retry_at, + rejection_count,deferred_at) VALUES (?,?,?,? ,?,?,?,?,?,?,?) ON CONFLICT(locator) DO UPDATE SET status=excluded.status, + error_code=excluded.error_code,error_message=excluded.error_message,retry_count=excluded.retry_count, + next_retry_at=excluded.next_retry_at,rejection_count=excluded.rejection_count,deferred_at=excluded.deferred_at`) + .run(locator, failureKind, "unknown", now, dead ? "dead-letter" : "retry", storedCode, storedMessage, attempts, next, + rejections, deferredAt); if (dead) { db.prepare(`INSERT INTO dead_letters(locator,kind,error_code,error_message,attempts,first_seen_at,last_seen_at) VALUES (?,?,?,?,?,?,?) ON CONFLICT(locator) DO UPDATE SET kind=excluded.kind,error_code=excluded.error_code, @@ -440,10 +601,33 @@ export function clearListingRejection(locator: string, registrationClaim: string db.prepare("DELETE FROM listing_rejections WHERE locator = ? AND registration_claim = ?") .run(locator, registrationClaim.toLowerCase()); } -export const loadRetryableArtifacts = (now = Date.now()): string[] => - (db.prepare("SELECT locator FROM artifacts WHERE (status='retry' OR error_code='ARTIFACT_REJECTED') AND COALESCE(next_retry_at,0) <= ? ORDER BY COALESCE(next_retry_at,0),locator LIMIT 100").all(now) as Array<{ locator: string }>).map((row) => row.locator); +/** + * Due retry locators for one pass, at most RETRY_READS_PER_PASS. Up to + * REJECTED_RETRY_READS_PER_PASS of them are reserved for due rejected + * locators; transient failures fill the rest and are returned first. Free + * scheduled slots are first given to rejected locators waiting beyond the + * ceiling, oldest first. + */ +export const loadRetryableArtifacts = db.transaction((now: number = Date.now()): string[] => { + const active = (countActiveRejected.get("") as { count: number }).count; + if (active < MAX_ACTIVE_REJECTED_RETRIES) { + db.prepare(`UPDATE artifacts SET deferred_at=NULL, next_retry_at=? WHERE locator IN (SELECT locator FROM artifacts + WHERE rejection_count > 0 AND deferred_at IS NOT NULL ORDER BY deferred_at, locator LIMIT ?)`) + .run(now, MAX_ACTIVE_REJECTED_RETRIES - active); + } + const due = (where: string, limit: number) => (db.prepare(`SELECT locator FROM artifacts WHERE ${where} + AND COALESCE(next_retry_at,0) <= ? ORDER BY COALESCE(next_retry_at,0),locator LIMIT ?`) + .all(now, limit) as Array<{ locator: string }>).map((row) => row.locator); + const rejected = due(REJECTED_ACTIVE, REJECTED_RETRY_READS_PER_PASS); + return [...due("status='retry' AND rejection_count = 0", RETRY_READS_PER_PASS - rejected.length), ...rejected]; +}); export const artifactAnchorTime = (locator: string): number | undefined => (db.prepare("SELECT anchor_time FROM artifacts WHERE locator = ?").get(locator) as { anchor_time: number | null } | undefined)?.anchor_time ?? undefined; +/** Canonical storage owner recorded when the locator was last read, if any. */ +export const artifactOwner = (locator: string): string | null => { + const owner = (db.prepare("SELECT owner FROM artifacts WHERE locator = ?").get(locator) as { owner: string | null } | undefined)?.owner; + return owner ? canonicalProgramOwner(owner) : null; +}; export function beginScanRun(fromTx: number): number { return Number(db.prepare("INSERT INTO scan_runs(started_at,from_tx,status) VALUES (?,?,?)").run(Date.now(), fromTx, "running").lastInsertRowid); diff --git a/reference-implementations/dacs-directory/src/catalog/types.ts b/reference-implementations/dacs-directory/src/catalog/types.ts index 5067de2..70f3a42 100644 --- a/reference-implementations/dacs-directory/src/catalog/types.ts +++ b/reference-implementations/dacs-directory/src/catalog/types.ts @@ -275,7 +275,7 @@ export interface ScanState { reachabilityCursor?: number; /** owner + programName → observed native address (nonce-safe binding). */ programs?: Record; - /** listing content hash → bounded, deterministic revocation candidates. */ + /** Legacy: listing content hash → revocation candidates. Moved into SQLite on load or save. */ revocations?: Record; /** RB-4-verified marker locators that candidate pruning must preserve. */ verifiedRevocations?: Record; @@ -285,6 +285,6 @@ export interface ScanState { bundleBindingOverflow?: string[]; /** listing anchor address → owner address */ listings: Record; - /** jobId → discovered deal */ + /** buyer owner + jobId (discoveredDealKey) → discovered deal */ deals: Record; } diff --git a/reference-implementations/dacs-directory/src/components/VerifyDeal.tsx b/reference-implementations/dacs-directory/src/components/VerifyDeal.tsx index 298d77c..17ffcb4 100644 --- a/reference-implementations/dacs-directory/src/components/VerifyDeal.tsx +++ b/reference-implementations/dacs-directory/src/components/VerifyDeal.tsx @@ -46,13 +46,13 @@ async function fetchArtifact(params: string): Promise | return json.value ?? null; } -async function registeredBundleRefs(jobId: string): Promise<{ +async function registeredBundleRefs(jobId: string, bundleRef: string): Promise<{ owners: unknown; buyerBundleRef: unknown; sellerBundleRef: unknown; } | null> { try { - const res = await fetch(`/api/dacs/deal-owners?jobId=${encodeURIComponent(jobId)}`); + const res = await fetch(`/api/dacs/deal-owners?jobId=${encodeURIComponent(jobId)}&bundleRef=${encodeURIComponent(bundleRef)}`); if (!res.ok) return null; return await res.json() as { owners: unknown; buyerBundleRef: unknown; sellerBundleRef: unknown }; } catch { @@ -129,7 +129,7 @@ export default function VerifyDeal({ const buyer = verification.bundle?.parties.find((p) => p.role === "buyer")?.primaryClaim; const seller = verification.bundle?.parties.find((p) => p.role === "seller")?.primaryClaim; const registered = verification.bundle?.jobId - ? await registeredBundleRefs(verification.bundle.jobId) : null; + ? await registeredBundleRefs(verification.bundle.jobId, bundleRef) : null; const anchorBindingAvailable = registeredAnchorRole( bundleRef, registered?.buyerBundleRef, diff --git a/reference-implementations/dacs-directory/test/bundle-binding.test.ts b/reference-implementations/dacs-directory/test/bundle-binding.test.ts index 615308d..4816730 100644 --- a/reference-implementations/dacs-directory/test/bundle-binding.test.ts +++ b/reference-implementations/dacs-directory/test/bundle-binding.test.ts @@ -131,7 +131,8 @@ test("BB-6 budget and the total discovery ceiling fail closed", async () => { const bounded = boundedBundleBindings(candidates, 8); assert.equal(bounded.bindings.length, 8); - assert.deepEqual(bounded.overflowKeys, [bundleBindingRoleKey(candidates[0].jobId, "buyer")]); + assert.equal(bounded.overflowKeys.length, 1); + assert.ok(bounded.overflowKeys[0].startsWith(bundleBindingRoleKey(candidates[0].jobId, "buyer"))); const overflow = await resolveBundleSide({ jobId: candidates[0].jobId, role: "buyer", @@ -143,6 +144,32 @@ test("BB-6 budget and the total discovery ceiling fail closed", async () => { assert.deepEqual(overflow, { disposition: "indeterminate", reason: "bundle-binding discovery cap exhausted" }); }); +test("the discovery ceiling is per signer: other signers' bindings leave the role holder's side intact", async () => { + const holder = await binding({ native: native(1), hash: "a".repeat(64) }); + const others = await Promise.all(Array.from({ length: 33 }, async (_, index) => { + const seed = Uint8Array.from(Buffer.alloc(32, 100 + index)); + const claim = `did:demos:agent:${Buffer.from(rawPublicKey(publicKeyFromSeed(seed))).toString("hex")}`; + const scope = { bindingVersion: "1" as const, jobId: holder.jobId, role: "buyer" as const, logicalAddress: holder.logicalAddress, + nativeAddress: native(100 + index), bundleContentHash: "b".repeat(64), signer: claim }; + const value = Buffer.from(await ed25519Sign(Buffer.from(`dacs-bundle-binding:v1:${contentHash(scope)}`, "utf8"), + privateKeyFromSeed(seed))).toString("base64url"); + return { ...scope, signature: { algorithm: "ed25519" as const, signer: claim, value } }; + })); + for (const other of others) assert.ok(await verifyBundleBinding(other)); + const bounded = boundedBundleBindings([...others, holder]); + assert.deepEqual(bounded.overflowKeys, []); + assert.equal(bounded.bindings.length, 34); + const resolution = await resolveBundleSide({ + jobId: holder.jobId, + role: "buyer", + expectedSigner: dids[0], + bindings: bounded.bindings, + overflow: bounded.overflowKeys.length > 0, + inspect: async (candidate) => ({ value: candidate, bundleContentHash: candidate.bundleContentHash, fullSignatureStanding: true }), + }); + assert.equal(resolution.disposition, "present"); +}); + test("BB-5 poisoned content hashes are inert and cannot win selection", async () => { const poisoned = await binding({ native: native(1), hash: "a".repeat(64) }); const honest = await binding({ native: native(2), hash: "b".repeat(64) }); diff --git a/reference-implementations/dacs-directory/test/current-indexer.test.ts b/reference-implementations/dacs-directory/test/current-indexer.test.ts index 28d7574..33949cb 100644 --- a/reference-implementations/dacs-directory/test/current-indexer.test.ts +++ b/reference-implementations/dacs-directory/test/current-indexer.test.ts @@ -9,7 +9,8 @@ import { deriveIdentityTier, type RecipePolicy } from "../src/catalog/identityVe import { indexRegistration, listingBindingRejection } from "../src/catalog/indexer.js"; import { deriveSellerReputation, isNeutralCancellation } from "../src/catalog/reputation.js"; import { verifyListing } from "../src/catalog/listingVerification.js"; -import { logicalBundleAddress } from "../src/catalog/bundleBinding.js"; +import { boundedBundleBindings, logicalBundleAddress } from "../src/catalog/bundleBinding.js"; +import { loadScanState, saveScanState } from "../src/catalog/store.js"; import type { BundleBinding, DealRecord, RegisteredDeal } from "../src/catalog/types.js"; type Obj = Record; @@ -813,6 +814,52 @@ test("indexer resolves current copies through BundleBindings rather than submitt } }); +test("bindings signed by other parties leave a deal's binding resolution intact", async () => { + const fixture = await vector("crowded-binding-job", 90); + const deal = registeredDeal("crowded-binding-job", locator(996), locator(997)); + const holders = [ + await bindingFor(deal.jobId, "buyer", fixture.locators.buyer, fixture.buyerBundle, 0), + await bindingFor(deal.jobId, "seller", fixture.locators.seller, fixture.sellerBundle, 1), + ]; + const others = await Promise.all(Array.from({ length: 33 }, async (_, index) => { + const seed = Uint8Array.from(Buffer.alloc(32, 150 + index)); + const claim = `did:demos:agent:${Buffer.from(rawPublicKey(publicKeyFromSeed(seed))).toString("hex")}`; + const scope = { bindingVersion: "1" as const, jobId: deal.jobId, role: "buyer" as const, logicalAddress: logicalBundleAddress(deal.jobId, "buyer"), + nativeAddress: locator(400 + index), bundleContentHash: "b".repeat(64), signer: claim }; + const value = Buffer.from(await ed25519Sign(Buffer.from(`dacs-bundle-binding:v1:${contentHash(scope)}`), privateKeyFromSeed(seed))).toString("base64url"); + return { ...scope, signature: { algorithm: "ed25519" as const, signer: claim, value } }; + })); + // Discovered bindings and their overflow, as a reindex pass would persist them, plus another + // signer's binding whose signature does not verify. + const discovered = boundedBundleBindings([...others, ...holders]); + const forged = { ...others[0], nativeAddress: locator(399) }; + const saved = loadScanState(); + saveScanState({ ...saved, bundleBindings: { ...saved.bundleBindings, [deal.jobId]: [...discovered.bindings, forged] }, + bundleBindingOverflow: [...saved.bundleBindingOverflow, ...discovered.overflowKeys] }); + const originalFetch = globalThis.fetch; + globalThis.fetch = async (input) => { + const data = maps.get(String(input).split("/").pop() ?? ""); + return new Response(JSON.stringify(data ? { success: true, owner: `0x${dids[1].slice(-64)}`, programName: "dacs:test", data } : { success: false }), + { status: data ? 200 : 404, headers: { "content-type": "application/json" } }); + }; + let omitted = 0; + const index = () => indexRegistration({ primaryClaim: dids[1], displayName: "seller", listingAnchors: [fixture.locators.listing], deals: [deal] }, + undefined, async () => { throw new Error("identity unavailable"); }, undefined, (kind, count) => { if (kind === "bindings") omitted += count; }); + try { + const record = await index(); + assert.equal(record.deals[0].refsVerified, true); + assert.equal(record.deals[0].buyerBundleRef, fixture.locators.buyer); + assert.equal(omitted, 0, "other signers' bindings are not verified for this deal"); + // Overflow of the role holder's own bindings still makes its side indeterminate. + const own = boundedBundleBindings([holders[0], await bindingFor(deal.jobId, "buyer", locator(995), fixture.buyerBundle, 0)], 1); + saveScanState({ ...loadScanState(), bundleBindingOverflow: own.overflowKeys }); + assert.equal((await index()).deals[0].refsVerified, false); + } finally { + globalThis.fetch = originalFetch; + saveScanState(saved); + } +}); + test("indexer excludes a valid current seller copy when the buyer anchor is unreadable", async () => { const fixture = await vector("seller-fallback-job", 80); maps.delete(fixture.locators.buyer); diff --git a/reference-implementations/dacs-directory/test/malformed-artifacts.test.ts b/reference-implementations/dacs-directory/test/malformed-artifacts.test.ts index f4b5d59..fa69174 100644 --- a/reference-implementations/dacs-directory/test/malformed-artifacts.test.ts +++ b/reference-implementations/dacs-directory/test/malformed-artifacts.test.ts @@ -151,7 +151,11 @@ test("reindex records a deeply nested rejected artifact as bounded metadata and const [indexed] = catalog.sellers[0].listings; assert.equal(indexed.status, "revoked"); assert.equal(indexed.revocationBinding?.markerAnchor.locator, validMarker); - assert.deepEqual(store.loadScanState().revocations?.[verifiedListing.contentHash], [validMarker]); + const queue = new Database(join(dataDirectory, "directory.sqlite"), { readonly: true }); + try { + assert.deepEqual(queue.prepare("SELECT locator FROM revocation_candidates WHERE listing_hash = ? AND state = 'pending'") + .all(verifiedListing.contentHash), [{ locator: validMarker }]); + } finally { queue.close(); } assert.deepEqual(artifactRow(deepMarker), { kind: "other", content_hash: null, anchor_time: null, data_json: null, status: "retry" }); assert.ok(store.loadRetryableArtifacts(Date.now() + 4_000_000).includes(deepMarker)); assert.equal(artifactRow(validMarker)?.content_hash, contentHash(marker)); @@ -216,9 +220,11 @@ test("scan requires an object payload and string metadata from every storage rea assert.equal(result.complete, true); assert.equal(result.listings.size, 0); - for (const invalid of [numeric, text, list, numericOwner]) { + // Non-object data under a DACS name is rejected content; a non-string owner is a malformed node response. + for (const [invalid, code] of [[numeric, "ARTIFACT_REJECTED"], [text, "ARTIFACT_REJECTED"], [list, "ARTIFACT_REJECTED"], + [numericOwner, "STORAGE_INVALID_RESPONSE"]]) { assert.equal(result.observations.some((observation) => observation.locator === invalid), false); - assert.ok(result.failures.some((failure) => failure.locator === invalid && failure.code === "STORAGE_INVALID_RESPONSE")); + assert.ok(result.failures.some((failure) => failure.locator === invalid && failure.code === code), invalid); } assert.deepEqual(result.revocations.get(verifiedListing.contentHash), [validMarker]); }); @@ -267,18 +273,19 @@ test("an unhashable program never shadows a valid agreement or supplies deal att return { data: member ? withMember(raw, member) : JSON.stringify(raw), name, owner: ownerOf(buyer) }; }; const key = store.programBindingKey(ownerOf(buyer), name); + const dealKey = `${ownerOf(buyer)}\njob-attribution`; chain({ [bundle]: buyerBundle, [agreement]: program(seller), [unhashable]: program(outsider, "1e400") }, [bundle, agreement, unhashable]); let result = await scan(); assert.equal(result.programs.get(key), agreement); - assert.equal(result.deals.get("job-attribution")?.owners.seller, seller); + assert.equal(result.deals.get(dealKey)?.owners.seller, seller); // Competing owner/name programs remain indeterminate whatever the read order. for (const order of [[bundle, agreement, otherAgreement], [bundle, otherAgreement, agreement]]) { chain({ [bundle]: buyerBundle, [agreement]: program(seller), [otherAgreement]: program(outsider) }, order); result = await scan(); assert.equal(result.programs.get(key), null); - assert.equal(result.deals.get("job-attribution")?.owners.seller, ""); + assert.equal(result.deals.get(dealKey)?.owners.seller, ""); } // The deal loop rereads the agreement; unhashable data on that read is rejected too. @@ -286,14 +293,14 @@ test("an unhashable program never shadows a valid agreement or supplies deal att reread = (at, count) => at === agreement && count > 1 ? withMember({ seller: outsider }, "1e400") : undefined; result = await scan(); assert.equal(result.programs.get(key), agreement); - assert.equal(result.deals.get("job-attribution")?.sellerFromAgreement, undefined); - assert.equal(result.deals.get("job-attribution")?.owners.seller, ""); + assert.equal(result.deals.get(dealKey)?.sellerFromAgreement, undefined); + assert.equal(result.deals.get(dealKey)?.owners.seller, ""); // A canonical agreement whose parties are not objects attributes nothing and stops nothing. chain({ [bundle]: buyerBundle, [agreement]: { data: JSON.stringify({ parties: [null, 7] }), name, owner: ownerOf(buyer) } }, [bundle, agreement]); result = await scan(); assert.equal(result.complete, true); - assert.equal(result.deals.get("job-attribution")?.owners.seller, ""); + assert.equal(result.deals.get(dealKey)?.owners.seller, ""); }); test("current evidence graph fails closed on malformed bundles without hashing them again", async () => { diff --git a/reference-implementations/dacs-directory/test/reindex-chain-reset.test.ts b/reference-implementations/dacs-directory/test/reindex-chain-reset.test.ts index 201dde2..caf11ec 100644 --- a/reference-implementations/dacs-directory/test/reindex-chain-reset.test.ts +++ b/reference-implementations/dacs-directory/test/reindex-chain-reset.test.ts @@ -55,7 +55,7 @@ test("reindex clears replaced-chain discoveries and rescans from genesis in one const result = await reindexAll({ log: (line) => logs.push(line) }); assert.equal(result.cursor, 3, "finality depth leaves the newest two transactions for replay"); - assert.equal(store.loadScanState().schemaVersion, 9); + assert.equal(store.loadScanState().schemaVersion, 10); assert.deepEqual(Object.keys(store.loadScanState().listings), []); assert.equal(store.indexerDiagnostics().deadLetters, 0); assert.match(logs[0] ?? "", /chain replacement detected/); diff --git a/reference-implementations/dacs-directory/test/retry-class-migration.test.ts b/reference-implementations/dacs-directory/test/retry-class-migration.test.ts new file mode 100644 index 0000000..0bba9f3 --- /dev/null +++ b/reference-implementations/dacs-directory/test/retry-class-migration.test.ts @@ -0,0 +1,69 @@ +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"; + +// A database written before rejected and transient retries were counted +// separately: one shared retry_count, and rejection dead letters from older +// releases. +const dataDirectory = mkdtempSync(join(tmpdir(), "dacs-directory-retry-migration-")); +const locator = (n: number) => `stor-${n.toString(16).padStart(40, "0")}`; +const [rejected, oldRejected, transient, exhausted] = [1, 2, 3, 4].map(locator); +const seeded = new Database(join(dataDirectory, "directory.sqlite")); +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), ('sr2-anchor-schema-version','2',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 + ); + CREATE TABLE dead_letters ( + locator TEXT PRIMARY KEY, kind TEXT NOT NULL, error_code TEXT NOT NULL, error_message TEXT NOT NULL, + attempts INTEGER NOT NULL, first_seen_at INTEGER NOT NULL, last_seen_at INTEGER NOT NULL + ); +`); +const insert = seeded.prepare(`INSERT INTO artifacts(locator,kind,profile,observed_at,status,error_code,error_message,retry_count,next_retry_at) + VALUES (?,?,?,1,?,?,?,?,?)`); +insert.run(rejected, "other", "unknown", "retry", "ARTIFACT_REJECTED", "rejected", 4, 5_000); +insert.run(oldRejected, "listing-revocation", "unknown", "dead-letter", "ARTIFACT_REJECTED", "rejected", 5, null); +insert.run(transient, "unknown", "unknown", "retry", "STORAGE_RPC_UNAVAILABLE", "outage", 2, 6_000); +insert.run(exhausted, "unknown", "unknown", "dead-letter", "STORAGE_RPC_UNAVAILABLE", "outage", 5, null); +// Enough further rejected rows to exceed the scheduled ceiling, due later than the rows above. +const later = Array.from({ length: 1_000 }, (_, index) => locator(10_000 + index)); +for (const [index, at] of later.entries()) insert.run(at, "other", "unknown", "retry", "ARTIFACT_REJECTED", "rejected", 1, 20_000 + index); +const deadLetter = seeded.prepare("INSERT INTO dead_letters VALUES (?,?,?,?,?,1,1)"); +deadLetter.run(oldRejected, "listing-revocation", "ARTIFACT_REJECTED", "rejected", 5); +deadLetter.run(exhausted, "unknown", "STORAGE_RPC_UNAVAILABLE", "outage", 5); +seeded.close(); + +process.env.DACS_DIRECTORY_DATA = dataDirectory; +const store = await import("../src/catalog/store.js"); + +test.after(() => rmSync(dataDirectory, { recursive: true, force: true })); + +test("existing retry rows are split into rejected and transient classes in place", () => { + assert.deepEqual(store.loadRetryableArtifacts(10_000), [transient, oldRejected, rejected], "transient retries first"); + const db = new Database(join(dataDirectory, "directory.sqlite"), { readonly: true }); + try { + const row = (at: string) => db.prepare("SELECT status, retry_count, rejection_count FROM artifacts WHERE locator=?").get(at); + assert.deepEqual(row(rejected), { status: "retry", retry_count: 0, rejection_count: 4 }); + assert.deepEqual(row(oldRejected), { status: "retry", retry_count: 0, rejection_count: 5 }); + assert.deepEqual(row(transient), { status: "retry", retry_count: 2, rejection_count: 0 }); + assert.deepEqual(row(exhausted), { status: "dead-letter", retry_count: 5, rejection_count: 0 }); + const deadLetters = db.prepare("SELECT locator FROM dead_letters ORDER BY locator").all(); + assert.deepEqual(deadLetters, [{ locator: exhausted }]); + } finally { db.close(); } +}); + +test("migrated rejected rows beyond the scheduled ceiling wait for a slot in due order", () => { + const db = new Database(join(dataDirectory, "directory.sqlite"), { readonly: true }); + try { + const count = (where: string) => (db.prepare(`SELECT COUNT(*) count FROM artifacts WHERE rejection_count > 0 AND ${where}`).get() as { count: number }).count; + assert.equal(count("deferred_at IS NULL"), 1_000); + const waiting = db.prepare("SELECT locator, next_retry_at FROM artifacts WHERE deferred_at IS NOT NULL ORDER BY locator").all(); + assert.deepEqual(waiting, later.slice(-2).map((locator) => ({ locator, next_retry_at: null }))); + } finally { db.close(); } +}); diff --git a/reference-implementations/dacs-directory/test/scan-fairness.test.ts b/reference-implementations/dacs-directory/test/scan-fairness.test.ts new file mode 100644 index 0000000..4dbdb25 --- /dev/null +++ b/reference-implementations/dacs-directory/test/scan-fairness.test.ts @@ -0,0 +1,809 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import Database from "better-sqlite3"; +import { NextRequest } from "next/server"; +import { contentHash } from "@kynesyslabs/dacs/canonical"; +import { ed25519Sign, privateKeyFromSeed, publicKeyFromSeed, rawPublicKey } from "@kynesyslabs/dacs/crypto"; +import type { BundleBinding, Registration, ScanState } from "../src/catalog/types.js"; + +// Each listing, deal and retry stays isolated from other parties' writes: +// real SDK hashes and signatures over a stubbed node, through reindexAll. +const dataDirectory = mkdtempSync(join(tmpdir(), "dacs-directory-fairness-")); +process.env.DACS_DIRECTORY_DATA = dataDirectory; +process.env.DACS_SCAN_FINALITY_DEPTH = "0"; +const store = await import("../src/catalog/store.js"); +const { verifyListing } = await import("../src/catalog/listingVerification.js"); +const { logicalBundleAddress, verifyBundleBinding } = await import("../src/catalog/bundleBinding.js"); +const { artifactHash } = await import("../src/catalog/evidenceGraph.js"); +const { indexRegistration } = await import("../src/catalog/indexer.js"); +const { deriveAnchorAddress, resolveOwnedAnchorByName } = await import("../src/catalog/chain.js"); +const { reindexAll } = await import("../src/catalog/reindexCore.js"); +const { scanChain } = await import("../src/catalog/scan.js"); +const { dedupeVerifiedDeals } = await import("../src/catalog/bundlePolicy.js"); +const artifactRoute = await import("../app/api/dacs/artifact/route.js"); +const dealOwnersRoute = await import("../app/api/dacs/deal-owners/route.js"); +const { registrationMessage } = await import("../src/catalog/registrationSig.js"); + +type Obj = Record; +const seeds = [21, 22, 23, 24].map((byte) => Uint8Array.from(Buffer.alloc(32, byte))); +const dids = seeds.map((seed) => `did:demos:agent:${Buffer.from(rawPublicKey(publicKeyFromSeed(seed))).toString("hex")}`); +const [buyer, seller, otherBuyer, otherSeller] = dids; +const ownerOf = (did: string) => `0x${did.slice(-64)}`; +const outsider = `did:demos:agent:${"d".repeat(64)}`; +const locator = (n: number) => `stor-${n.toString(16).padStart(40, "0")}`; +const range = (from: number, count: number) => Array.from({ length: count }, (_, index) => locator(from + index)); +const offline = async (): Promise => { throw new Error("identity unavailable"); }; +const signature = async (hash: string, prefix: string, signer: number, encoding: "hex" | "base64url" = "hex") => + Buffer.from(await ed25519Sign(Buffer.from(prefix + hash), privateKeyFromSeed(seeds[signer]))).toString(encoding); + +const listingScope = { + listingId: "svc", listingVersion: 1, agentId: seller, serviceId: "svc", name: "Service", description: "Description", + claimRequirements: [], supportedNegotiation: ["negotiate-fixed-price"], supportedPaymentRails: ["pay-dem"], + supportedDelivery: ["deliver-attested-payload"], +}; +const listing = { ...listingScope, signature: { algorithm: "ed25519", signer: seller, value: await signature(contentHash(listingScope), "dacs-listing:v1:", 1) } }; +const verifiedListing = await verifyListing(listing); +assert.ok(verifiedListing); +const listingHash = verifiedListing.contentHash; +const markerScope = { listingId: "svc", listingVersion: 1, listingContentHash: listingHash, revokedAt: 1 }; +const marker = { ...markerScope, signature: { algorithm: "ed25519", signer: seller, value: await signature(contentHash(markerScope), "dacs-revocation:v1:", 1) } }; +/** Marker-shaped, but its signature does not verify. */ +const unsigned = (n: number) => ({ ...markerScope, revokedAt: 1_000 + n, signature: { algorithm: "ed25519", signer: seller, value: "00".repeat(64) } }); +async function agreementFor(jobId: string, buyerIndex: number, sellerIndex: number) { + const scope = { agreementVersion: "1", jobId, listingRef: { listingId: "svc", version: 1, contentHash: listingHash }, + parties: [{ role: "buyer", primaryClaim: dids[buyerIndex] }, { role: "seller", primaryClaim: dids[sellerIndex] }], + terms: { price: { amount: "1.25", currency: "DEM" }, rail: { railId: "pay-dem" } } }; + const sign = async (index: number) => ({ party: dids[index], algorithm: "ed25519", + value: await signature(artifactHash(scope, "agreement"), "dacs-agreement:v1:", index) }); + return { ...scope, signatures: [await sign(buyerIndex), await sign(sellerIndex)] }; +} + +// Stubbed node. `unavailable` locators answer 503; `scanOnly` locators answer only their first read. +interface StoredProgram { data: string; name?: string; owner?: string } +let storage: Record = {}; +let transactions: Obj[] = []; +let reads = new Map(); +const unavailable = new Set(); +const scanOnly = new Set(); +let nodeDown = false; +globalThis.fetch = async (input, init) => { + const path = new URL(String(input)).pathname; + if (path.startsWith("/storage-program/")) { + const at = path.slice("/storage-program/".length); + const count = (reads.get(at) ?? 0) + 1; + reads.set(at, count); + const entry = storage[at]; + if (!entry) return new Response(null, { status: 404 }); + if (unavailable.has(at) || (scanOnly.has(at) && count > 1)) return new Response(null, { status: 503 }); + return new Response(`{"success":true,"owner":${JSON.stringify(entry.owner ?? ownerOf(seller))},` + + `"programName":${JSON.stringify(entry.name ?? "opaque")},"data":${entry.data}}`, { headers: { "content-type": "application/json" } }); + } + const call = JSON.parse(String(init?.body)).params?.[0]; + if (nodeDown) return Response.json({ result: 503 }); + return Response.json({ result: 200, response: call?.message === "getTransactions" && call.data?.start === "latest" ? transactions : [] }); +}; +const memo = (id: number, at?: string): Obj => ({ id, status: "confirmed", type: "transfer", hash: id.toString(16).padStart(64, "0"), + blockNumber: id, to: "0x1", content: JSON.stringify(at ? { memo: at } : {}) }); +/** `written` is in chain order (oldest first) from tx id `firstId`; the node returns newest first. */ +function chain(programs: Record, written: string[], firstId = 1) { + storage = programs; + reads = new Map(); + unavailable.clear(); + scanOnly.clear(); + transactions = written.map((at, index) => memo(firstId + index, at)).reverse(); +} +const readsOf = (locators: string[]) => locators.reduce((total, at) => total + (reads.get(at) ?? 0), 0); + +function sql(query: string, ...params: unknown[]) { + const db = new Database(join(dataDirectory, "directory.sqlite")); + try { + const statement = db.prepare(query); + return (statement.reader ? statement.all(...params) : (statement.run(...params), [])) as Record[]; + } finally { db.close(); } +} +const emptyState = (): ScanState => ({ + schemaVersion: 9, lastSeenTxId: 0, listings: {}, deals: {}, programs: {}, revocations: {}, + verifiedRevocations: {}, bundleBindings: {}, bundleBindingOverflow: [], anchorBackfillComplete: true, +}); +const candidateTable = () => sql("SELECT name FROM sqlite_master WHERE name='revocation_candidates'").length > 0; +function resetIndex(registrations: Registration[], generatedAt: number) { + store.saveRegistrations(registrations); + store.saveCatalog({ catalogVersion: "1", generatedAt, sellers: [] }); + store.saveScanState(emptyState()); + sql("DELETE FROM artifacts"); + sql("DELETE FROM dead_letters"); + if (candidateTable()) sql("DELETE FROM revocation_candidates"); +} +/** Candidates still queued for a listing, in the order they will be examined. */ +const pending = (hash: string) => sql("SELECT locator FROM revocation_candidates WHERE listing_hash=? AND state='pending' ORDER BY seq", hash) + .map((row) => String(row.locator)); +/** SQLite statements executed, and rows they returned, in this process while enabled. */ +const sqlWork = { enabled: false, calls: 0, rows: 0 }; +{ + const probe = new Database(":memory:"); + const statement = Object.getPrototypeOf(probe.prepare("SELECT 1")) as Record unknown>; + probe.close(); + for (const method of ["get", "all", "run", "iterate"]) { + const original = statement[method]; + statement[method] = function (this: unknown, ...args: unknown[]) { + const result = original.apply(this, args); + if (sqlWork.enabled) { + sqlWork.calls++; + sqlWork.rows += Array.isArray(result) ? result.length : result === undefined ? 0 : 1; + } + return result; + }; + } +} +const later: Registration = { primaryClaim: outsider, displayName: "Later", listingAnchors: [] }; +const sellerRegistration: Registration = { primaryClaim: seller, displayName: "Seller", listingAnchors: [locator(1)] }; +const reindex = () => reindexAll({ log: () => {}, resolveIdentities: offline }); +const listingStatus = () => store.loadCatalog().sellers.find((s) => s.primaryClaim === seller)?.listings[0]?.status; +const lastRun = () => sql("SELECT * FROM scan_runs ORDER BY id DESC LIMIT 1")[0]; + +test.after(() => rmSync(dataDirectory, { recursive: true, force: true })); + +test("a revocation anchored by the listing owner is applied even when many other candidates exist", async () => { + resetIndex([sellerRegistration, later], 5); + const older = range(100, 20), newer = range(150, 20); + chain({ + [locator(1)]: { data: JSON.stringify(listing) }, + [locator(2)]: { data: JSON.stringify(marker) }, + ...Object.fromEntries([...older, ...newer].map((at, n) => [at, { data: JSON.stringify(unsigned(n)), owner: ownerOf(outsider) }])), + }, [locator(1), ...older, locator(2), ...newer]); + await reindex(); + assert.equal(listingStatus(), "revoked", "verified in the first pass"); + const candidates = [locator(2), ...older, ...newer]; + // The scanner reads each new locator once; the rest is verification work. + assert.ok(readsOf(candidates) - candidates.length <= 16, `pass 1 verified ${readsOf(candidates) - candidates.length}`); + transactions = []; + reads = new Map(); + await reindex(); + assert.equal(listingStatus(), "revoked"); + assert.ok(readsOf(candidates) <= 16); +}); + +test("every revocation candidate is eventually verified with bounded work per pass", async () => { + resetIndex([sellerRegistration, later], 6); + // The valid marker is anchored by a third party, so it gets no ordering priority. + // Ahead of it: 16 candidates unreadable after discovery, then 20 that fail verification. + const older = range(200, 36), newer = range(300, 20); + const unreadable = older.slice(0, 16); + chain({ + [locator(1)]: { data: JSON.stringify(listing) }, + [locator(3)]: { data: JSON.stringify(marker), owner: ownerOf(otherBuyer) }, + ...Object.fromEntries([...older, ...newer].map((at, n) => [at, { data: JSON.stringify(unsigned(n)), owner: ownerOf(outsider) }])), + }, [locator(1), ...older, locator(3), ...newer]); + for (const at of unreadable) scanOnly.add(at); + const candidates = [...older, locator(3), ...newer]; + await reindex(); + assert.ok(readsOf(candidates) - candidates.length <= 16, `pass 1 verified ${readsOf(candidates) - candidates.length}`); + let passes = 1; + transactions = []; + while (listingStatus() !== "revoked" && passes < 6) { + reads = new Map(); + await reindex(); + passes++; + assert.ok(readsOf(candidates) <= 16, `pass ${passes} verified ${readsOf(candidates)}`); + } + assert.equal(listingStatus(), "revoked"); + assert.ok(passes <= Math.ceil(candidates.length / 16), `revoked after ${passes} passes`); + // Unreadable candidates are kept for a later pass rather than dropped. + const queued = pending(listingHash); + for (const at of unreadable) assert.ok(queued.includes(at), `${at} stays queued`); + // Rejected candidates are not verified again. + reads = new Map(); + await reindex(); + assert.equal(listingStatus(), "revoked"); + assert.equal(readsOf(older.slice(16)), 0); + assert.ok(readsOf(candidates) <= 16); +}); + +test("per-pass revocation work does not grow with the number of queued candidates", async () => { + const work: Array<{ calls: number; rows: number; verified: number }> = []; + for (const count of [40, 400]) { + resetIndex([sellerRegistration, later], 17); + const queued = range(20_000, count); + chain({ + [locator(1)]: { data: JSON.stringify(listing) }, + ...Object.fromEntries(queued.map((at, n) => [at, { data: JSON.stringify(unsigned(n)), owner: ownerOf(outsider) }])), + }, [locator(1), ...queued]); + await reindex(); + transactions = []; + reads = new Map(); + sqlWork.calls = sqlWork.rows = 0; + sqlWork.enabled = true; + try { await reindex(); } finally { sqlWork.enabled = false; } + work.push({ calls: sqlWork.calls, rows: sqlWork.rows, verified: readsOf(queued) }); + } + assert.deepEqual(work[1], work[0], "the same statements and rows with ten times the queue"); + assert.equal(work[0].verified, 16); + // Nothing unexamined left the queue: two passes examined 32 of the 400. + assert.equal(pending(listingHash).length, 400 - 32); +}); + +test("a rejected candidate is verified again only when its content changes", async () => { + resetIndex([sellerRegistration, later], 18); + const candidates = range(600, 3); + chain({ + [locator(1)]: { data: JSON.stringify(listing) }, + ...Object.fromEntries(candidates.map((at, n) => [at, { data: JSON.stringify(unsigned(n)), owner: ownerOf(outsider) }])), + }, [locator(1), ...candidates]); + await reindex(); + assert.equal(listingStatus(), "active"); + // Mentioned again with the same content: the scanner reads each once and none is verified again. + transactions = [memo(50, candidates.join(" "))]; + reads = new Map(); + await reindex(); + assert.equal(readsOf(candidates), candidates.length); + // New content at one of them is examined. + storage[candidates[1]] = { data: JSON.stringify(marker), owner: ownerOf(outsider) }; + transactions = [memo(60, candidates[1]), ...transactions]; + await reindex(); + assert.equal(listingStatus(), "revoked"); +}); + +test("a verified marker missing from the candidate queue is still checked first", async () => { + resetIndex([sellerRegistration, later], 15); + const others = range(400, 20); + chain({ + [locator(1)]: { data: JSON.stringify(listing) }, + [locator(4)]: { data: JSON.stringify(marker), owner: ownerOf(otherBuyer) }, + ...Object.fromEntries(others.map((at, n) => [at, { data: JSON.stringify(unsigned(n)), owner: ownerOf(outsider) }])), + }, [locator(1)]); + // A snapshot that kept the verified locator but not its queue entry. + store.saveScanState({ ...emptyState(), revocations: { [listingHash]: others }, verifiedRevocations: { [listingHash]: [locator(4)] } }); + await reindex(); + assert.equal(listingStatus(), "revoked"); + assert.ok(readsOf(others) === 0, "the verified marker is checked before any other candidate"); +}); + +for (const route of ["unsigned bundle-named program", "bundle and agreement signed by its own parties"]) { + test(`deals are keyed by their owner: another owner's ${route} with the same jobId leaves the deal in place`, async () => { + const jobId = "shared-job"; + const signedRoute = !route.startsWith("unsigned"); + // The signed route uses current-format bundles; the unsigned route uses name-only programs. + const currentBundle = JSON.stringify({ bundleVersion: "1", jobId, parties: [], anchoredByRole: "buyer" }); + resetIndex([later], 7); + chain({ + [locator(40)]: { name: `dacs5:bundle:${jobId}`, data: signedRoute ? currentBundle : '{"any":1}', owner: ownerOf(buyer) }, + [locator(41)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + }, [locator(40), locator(41)]); + transactions.unshift(memo(10)); + await reindex(); + const dealsFor = (did: string) => store.loadCatalog().sellers.find((s) => s.primaryClaim === did)?.deals ?? []; + const entryOf = (did: string) => Object.values(store.loadScanState().deals).filter((deal) => deal.owners.buyer === did && deal.jobId === jobId); + assert.deepEqual(dealsFor(seller).map((deal) => deal.buyerBundleRef), [locator(40)]); + + // The established writes are outside the next pass's replay overlap. + const writer = signedRoute ? otherBuyer : outsider; + storage[locator(42)] = signedRoute + ? { data: currentBundle, owner: ownerOf(writer) } + : { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(writer) }; + const second = [locator(42)]; + if (signedRoute) { + storage[locator(43)] = { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 2, 3)), owner: ownerOf(otherBuyer) }; + second.push(locator(43)); + } + transactions = [...second.map((at, index) => memo(20 + index, at)).reverse(), ...transactions]; + for (let pass = 2; pass <= 3; pass++) { + reads = new Map(); + await reindex(); + assert.ok(Number(lastRun().from_tx) > 2, "the established writes (tx 1 and 2) are outside the replay overlap"); + assert.equal(entryOf(buyer).length, 1, `pass ${pass}`); + assert.equal(entryOf(buyer)[0].owners.seller, seller); + assert.deepEqual(dealsFor(seller).map((deal) => deal.buyerBundleRef), [locator(40)], `pass ${pass}`); + assert.equal(entryOf(writer).length, 1, "the other owner's program is its own entry"); + if (!signedRoute) assert.equal(entryOf(writer)[0].owners.seller, ""); + else assert.deepEqual(dealsFor(otherSeller).map((deal) => deal.buyerBundleRef), [locator(42)]); + } + // Within one scan window as well, each owner's program is its own entry. + transactions = [...second, locator(41), locator(40)].map((at, index) => memo(40 - index, at)); + const window = await scanChain(null, { maxTxs: 100, sinceTxId: 0 }); + const scanned = [...window.deals.values()].filter((deal) => deal.jobId === jobId); + assert.deepEqual(scanned.map((deal) => deal.owners.buyer).sort(), [buyer, writer].sort()); + assert.equal(scanned.find((deal) => deal.owners.buyer === buyer)?.owners.seller, seller); + }); +} + +test("two buyers' attributed deals with the same jobId both reach the seller's record", async () => { + const jobId = "two-buyers"; + resetIndex([later], 19); + chain({ + [locator(70)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, + [locator(71)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + [locator(72)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":2}', owner: ownerOf(otherBuyer) }, + [locator(73)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 2, 1)), owner: ownerOf(otherBuyer) }, + }, [locator(70), locator(71), locator(72), locator(73)]); + await reindex(); + const deals = store.loadCatalog().sellers.find((s) => s.primaryClaim === seller)?.deals ?? []; + assert.deepEqual(deals.map((deal) => `${deal.owners.buyer} ${deal.buyerBundleRef}`).sort(), + [`${buyer} ${locator(70)}`, `${otherBuyer} ${locator(72)}`].sort()); + // Verified-deal deduplication keeps one record per buyer and jobId, and per bundle ref. + const verifiedDeal = (who: string, at: number) => ({ jobId, rail: "pay-dem", buyerBundleRef: locator(at), + owners: { buyer: who, seller }, signatureVerified: true, refsVerified: true, verifiedAt: 1 }); + assert.deepEqual(dedupeVerifiedDeals([verifiedDeal(buyer, 70), verifiedDeal(otherBuyer, 72), verifiedDeal(buyer, 74), verifiedDeal(outsider, 70)]) + .map((deal) => deal.buyerBundleRef), [locator(70), locator(72)]); +}); + +test("v9 scan state is completed by a full deal-history replay before it is marked migrated", async () => { + const jobId = "stored-job"; + resetIndex([later], 8); + chain({ + [locator(45)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, + [locator(46)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + [locator(47)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":2}', owner: ownerOf(otherBuyer) }, + [locator(48)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 2, 3)), owner: ownerOf(otherBuyer) }, + }, [locator(45), locator(46), locator(47), locator(48)]); + transactions.unshift(memo(30)); + // jobId-keyed v9 state in which only one owner's entry was kept. + const kept = { jobId, rail: "pay-dem", buyerBundleRef: locator(47), owners: { buyer: otherBuyer, seller: otherSeller }, sellerFromAgreement: otherSeller }; + sql("UPDATE kv_state SET value_json=? WHERE key='scan-state'", JSON.stringify({ ...emptyState(), lastSeenTxId: 30, deals: { [jobId]: kept } })); + const entries = () => Object.values(store.loadScanState().deals).filter((deal) => deal.jobId === jobId).map((deal) => deal.owners.buyer).sort(); + // An interrupted replay leaves the state unmigrated, and the next pass replays again. + nodeDown = true; + try { await assert.rejects(reindex()); } finally { nodeDown = false; } + assert.equal(store.loadScanState().schemaVersion, 9); + assert.deepEqual(entries(), [otherBuyer]); + // An entry whose bundle cannot be read during the replay is kept, not dropped. + unavailable.add(locator(47)); + await reindex(); + unavailable.clear(); + assert.equal(lastRun().from_tx, 0, "a replay from genesis"); + assert.equal(store.loadScanState().schemaVersion, 10); + assert.deepEqual(entries(), [buyer, otherBuyer].sort()); + assert.deepEqual(store.loadCatalog().sellers.find((s) => s.primaryClaim === seller)?.deals.map((d) => d.buyerBundleRef), [locator(45)]); + // Once migrated, passes are incremental and the result is unchanged. + await reindex(); + assert.equal(lastRun().from_tx, 28); + assert.deepEqual(entries(), [buyer, otherBuyer].sort()); + const saved = JSON.parse(String(sql("SELECT value_json FROM kv_state WHERE key='scan-state'")[0].value_json)) as ScanState; + assert.deepEqual(Object.keys(saved.deals).sort(), [`${ownerOf(buyer)}\n${jobId}`, `${ownerOf(otherBuyer)}\n${jobId}`].sort()); +}); + +test("a discovered deal keeps its seller copy when a later window has none", async () => { + const jobId = "kept-copy"; + resetIndex([later], 23); + chain({ + [locator(920)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, + [locator(921)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + [locator(922)]: { name: `dacs5:bundle:seller:${jobId}`, data: '{"any":1}', owner: ownerOf(seller) }, + }, [locator(920), locator(921), locator(922)]); + transactions.unshift(memo(10)); + await reindex(); + const entry = () => store.loadScanState().deals[`${ownerOf(buyer)}\n${jobId}`]; + assert.equal(entry().sellerBundleRef, locator(922)); + transactions = [memo(40, locator(920))]; + await reindex(); + assert.equal(entry().owners.seller, seller); + assert.equal(entry().sellerBundleRef, locator(922)); +}); + +/** Records each locator as read with `who` as its storage owner. */ +const owns = (who: string, ...at: number[]) => { + for (const n of at) store.recordArtifact({ locator: locator(n), kind: "bundle", profile: "unknown", owner: ownerOf(who), observedAt: 1 }); +}; +const ownersLookup = (jobId: string) => async (query = "") => + await (await dealOwnersRoute.GET(new NextRequest(`http://localhost/api/dacs/deal-owners?jobId=${jobId}${query}`))).json(); + +test("deal-owner lookup answers only when one attributed deal holds the jobId", async () => { + const jobId = "lookup-job"; + const entry = (who: string, sellerClaim: string, at: number) => + ({ jobId, rail: "pay-dem", buyerBundleRef: locator(at), owners: { buyer: who, seller: sellerClaim } }); + const lookup = async (query = "") => + await (await dealOwnersRoute.GET(new NextRequest(`http://localhost/api/dacs/deal-owners?jobId=${jobId}${query}`))).json(); + const verified = (deal: ReturnType) => ({ ...deal, refsVerified: true }); + const catalogOf = (...records: Array<[string, Array & { refsVerified?: boolean; sellerBundleRef?: string }>]>) => store.saveCatalog({ catalogVersion: "1", generatedAt: 1, + sellers: records.map(([claim, deals]) => ({ primaryClaim: claim, + deals: deals.map((deal) => ({ signatureVerified: false, refsVerified: false, verifiedAt: 1, ...deal })) })) as never }); + resetIndex([later], 14); + owns(buyer, 46, 56); + owns(outsider, 47); + owns(otherBuyer, 48); + // Scanned entries only: an attributed entry next to another owner's unattributed one is ambiguous. + store.saveScanState({ ...emptyState(), deals: { a: entry(buyer, seller, 46), b: entry(outsider, "", 47) } }); + assert.equal((await lookup()).owners, null); + store.saveScanState({ ...emptyState(), deals: { a: entry(buyer, seller, 46) } }); + assert.deepEqual((await lookup()).owners, { buyer, seller }); + // A lone unattributed entry is not reported as ownership. + store.saveScanState({ ...emptyState(), deals: { b: entry(outsider, "", 47) } }); + assert.deepEqual(await lookup(), { owners: null, buyerBundleRef: null }); + // A verified catalog entry stands for its own scanned entry, whose refs may differ. + store.saveScanState({ ...emptyState(), deals: { a: entry(buyer, seller, 46) } }); + catalogOf([seller, [verified(entry(buyer, seller, 56))]]); + assert.deepEqual(await lookup(), { owners: { buyer, seller }, buyerBundleRef: locator(56), sellerBundleRef: null }); + // An unverified one does not when its ref is neither the scanned ref nor bound by a binding. + const scannedDeal = { owners: { buyer, seller }, buyerBundleRef: locator(46), sellerBundleRef: null }; + catalogOf([seller, [entry(buyer, seller, 56)]]); + assert.deepEqual(await lookup(), scannedDeal); + // Nor does a verified one with a ref its party does not hold, or one in another seller's record. + catalogOf([seller, [verified(entry(buyer, seller, 47))]]); + assert.deepEqual(await lookup(), scannedDeal); + catalogOf([seller, [{ ...verified(entry(buyer, seller, 46)), sellerBundleRef: locator(47) }]]); + assert.deepEqual(await lookup(), scannedDeal); + catalogOf([otherSeller, [verified(entry(buyer, seller, 56))]]); + assert.deepEqual(await lookup(), scannedDeal); + // Nor an unverified one naming the scanned buyer's ref under another seller. + catalogOf([otherSeller, [entry(buyer, otherSeller, 46)]]); + assert.deepEqual(await lookup(), scannedDeal); + // Two catalog sellers' deals with the jobId are ambiguous unless the bundle names one. + catalogOf([seller, [entry(buyer, seller, 46)]], [otherSeller, [verified(entry(otherBuyer, otherSeller, 48))]]); + assert.equal((await lookup()).owners, null); + assert.deepEqual((await lookup(`&bundleRef=${locator(48)}`)).owners, { buyer: otherBuyer, seller: otherSeller }); +}); + +test("a seller copy belongs only to a deal whose attributed seller owns it", async () => { + const jobId = "copy-owner"; + resetIndex([later], 24); + chain({ + [locator(930)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, + [locator(931)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + [locator(932)]: { name: `dacs5:bundle:seller:${jobId}`, data: '{"any":1}', owner: ownerOf(seller) }, + }, [locator(930), locator(931), locator(932)]); + transactions.unshift(memo(10)); + await reindex(); + const lookup = ownersLookup(jobId); + const deal = { owners: { buyer, seller }, buyerBundleRef: locator(930), sellerBundleRef: locator(932) }; + assert.deepEqual(await lookup(`&bundleRef=${locator(932)}`), deal); + // Another owner's name-only bundle for the jobId, in the same window as the seller copy. + storage[locator(933)] = { name: `dacs5:bundle:${jobId}`, data: '{"any":9}', owner: ownerOf(outsider) }; + transactions = [memo(20, `${locator(933)} ${locator(932)}`), ...transactions]; + const entryOf = (did: string) => store.loadScanState().deals[`${ownerOf(did)}\n${jobId}`]; + const window = await scanChain(null, { maxTxs: 100, sinceTxId: 0 }); + const scannedOf = (did: string) => [...window.deals.values()].find((deal) => deal.jobId === jobId && deal.owners.buyer === did); + assert.equal(scannedOf(outsider)?.sellerBundleRef, undefined); + assert.equal(scannedOf(buyer)?.sellerBundleRef, locator(932)); + for (let pass = 1; pass <= 3; pass++) { + await reindex(); + transactions = []; + assert.equal(entryOf(outsider).owners.seller, ""); + assert.equal(entryOf(outsider).sellerBundleRef, undefined, `pass ${pass}`); + assert.deepEqual(await lookup(`&bundleRef=${locator(932)}`), deal, `pass ${pass}`); + assert.deepEqual(await lookup(`&bundleRef=${locator(930)}`), deal, `pass ${pass}`); + } + // State saved by an earlier release can still carry a copy on the unattributed entry: + // the lookup does not count it, and the next pass removes it. + const unfiltered = await lookup(); + const state = store.loadScanState(); + state.deals[`${ownerOf(outsider)}\n${jobId}`].sellerBundleRef = locator(932); + store.saveScanState(state); + assert.deepEqual(await lookup(`&bundleRef=${locator(932)}`), deal); + assert.deepEqual(await lookup(), unfiltered); + await reindex(); + assert.equal(entryOf(outsider).sellerBundleRef, undefined); +}); + +test("deal-owner lookup counts a deal only when its bundle refs are bound to its parties", async () => { + const jobId = "declared-job"; + const declared = (buyerRef: number, sellerRef: number, owners = { buyer, seller }) => + ({ jobId, rail: "pay-dem", buyerBundleRef: locator(buyerRef), sellerBundleRef: locator(sellerRef), owners }); + const thirdParty: Registration = { primaryClaim: otherSeller, displayName: "Third party", listingAnchors: [], deals: [declared(943, 944)] }; + resetIndex([thirdParty, later], 25); + chain({ + [locator(940)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, + [locator(941)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + [locator(942)]: { name: `dacs5:bundle:seller:${jobId}`, data: '{"any":1}', owner: ownerOf(seller) }, + [locator(943)]: { data: '{"any":2}', owner: ownerOf(otherSeller) }, + [locator(944)]: { data: '{"any":3}', owner: ownerOf(otherSeller) }, + }, [locator(940), locator(941), locator(942), locator(943), locator(944)]); + transactions.unshift(memo(10)); + await reindex(); + const lookup = ownersLookup(jobId); + const deal = { owners: { buyer, seller }, buyerBundleRef: locator(940), sellerBundleRef: locator(942) }; + const catalogRefs = () => store.loadCatalog().sellers.flatMap((s) => s.deals).filter((d) => d.jobId === jobId).map((d) => d.buyerBundleRef).sort(); + assert.deepEqual(catalogRefs(), [locator(940), locator(943)], "the registration's deal is in the catalog"); + // The registration's refs are owned by neither named party. + assert.deepEqual(await lookup(`&bundleRef=${locator(944)}`), { owners: null, buyerBundleRef: null }); + assert.deepEqual(await lookup(`&bundleRef=${locator(940)}`), deal); + assert.deepEqual(await lookup(`&bundleRef=${locator(942)}`), deal); + assert.deepEqual(await lookup(), deal); + // A registration that names the buyer's bundle under other parties is not counted either. + store.saveRegistrations([{ ...thirdParty, deals: [declared(940, 944, { buyer: otherBuyer, seller: otherSeller })] }, later]); + transactions = []; + await reindex(); + assert.deepEqual(await lookup(`&bundleRef=${locator(940)}`), deal); + // Nor one that pairs the buyer's bundle with a seller copy the seller does not own. + store.saveRegistrations([{ ...thirdParty, deals: [declared(940, 944)] }, later]); + await reindex(); + assert.ok(catalogRefs().filter((ref) => ref === locator(940)).length === 2, "both deals are in the catalog"); + assert.deepEqual(await lookup(`&bundleRef=${locator(944)}`), { owners: null, buyerBundleRef: null }); + assert.deepEqual(await lookup(`&bundleRef=${locator(940)}`), deal); +}); + +test("a registration's deal counts for the lookup only when verified or when it is the scanned deal as indexed", async () => { + const jobId = "registered-job"; + const real = { owners: { buyer, seller }, buyerBundleRef: locator(960), sellerBundleRef: locator(962) }; + const declared = (buyerRef: number, sellerRef?: number, owners = { buyer, seller }) => ({ jobId, rail: "pay-dem", + buyerBundleRef: locator(buyerRef), ...(sellerRef ? { sellerBundleRef: locator(sellerRef) } : {}), owners }); + const signedBy = async (reg: Registration, signer: number): Promise => { + const signedAt = Date.now(); + const message = registrationMessage(reg, signedAt); + const value = Buffer.from(await ed25519Sign(Buffer.from(message), privateKeyFromSeed(seeds[signer]))).toString("hex"); + return { ...reg, ownerSignature: { message, signature: value, signedAt } }; + }; + const lookupsWith = async (registration: Registration) => { + resetIndex([registration, later], 27); + chain({ + [locator(960)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, + [locator(961)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify(await agreementFor(jobId, 0, 1)), owner: ownerOf(buyer) }, + [locator(962)]: { name: `dacs5:bundle:seller:${jobId}`, data: '{"any":1}', owner: ownerOf(seller) }, + }, [locator(960), locator(961), locator(962)]); + transactions.unshift(memo(10)); + await reindex(); + const lookup = ownersLookup(jobId); + return [await lookup(`&bundleRef=${locator(960)}`), await lookup(`&bundleRef=${locator(962)}`), await lookup()]; + }; + const thirdParty = (deal: ReturnType): Registration => + ({ primaryClaim: otherSeller, displayName: "Third party", listingAnchors: [], deals: [deal] }); + // Another party's registration naming the scanned deal's parties or refs, signed by that party or not. + for (const registration of [ + thirdParty(declared(960)), + thirdParty(declared(961, 962)), + thirdParty(declared(960, undefined, { buyer, seller: otherSeller })), + await signedBy(thirdParty(declared(960, undefined, { buyer, seller: otherSeller })), 3), + ]) { + assert.deepEqual(await lookupsWith(registration), [real, real, real], JSON.stringify(registration.deals)); + } + assert.equal(store.loadCatalog().sellers.find((s) => s.primaryClaim === otherSeller)?.ownerRegistered, true); + // An unsigned registration under the seller's own claim whose refs differ from the scanned deal's. + for (const deal of [declared(961, 962), declared(960)]) { + assert.deepEqual(await lookupsWith({ primaryClaim: seller, displayName: "Seller", listingAnchors: [], deals: [deal] }), + [real, real, real], JSON.stringify(deal)); + } +}); + +test("deal-owner lookup by bundleRef never reports a scanned ref the catalog has resolved past", async () => { + const jobId = "resolved-ref"; + resetIndex([later], 26); + const scanned = { jobId, rail: "pay-dem", buyerBundleRef: locator(950), owners: { buyer, seller } }; + owns(buyer, 950); + // The catalog's buyer ref comes from the buyer's verified BundleBinding. + const scope = { bindingVersion: "1", jobId, role: "buyer", logicalAddress: logicalBundleAddress(jobId, "buyer"), + nativeAddress: locator(951), bundleContentHash: "c".repeat(64), signer: buyer }; + const binding = { ...scope, signature: { algorithm: "ed25519", signer: buyer, + value: await signature(contentHash(scope), "dacs-bundle-binding:v1:", 0, "base64url") } } as BundleBinding; + assert.ok(await verifyBundleBinding(binding)); + store.saveScanState({ ...emptyState(), schemaVersion: 10, deals: { [`${ownerOf(buyer)}\n${jobId}`]: scanned }, bundleBindings: { [jobId]: [binding] } }); + store.saveCatalog({ catalogVersion: "1", generatedAt: 1, sellers: [{ primaryClaim: seller, + deals: [{ ...scanned, buyerBundleRef: locator(951), signatureVerified: true, refsVerified: true, verifiedAt: 1 }] }] } as never); + const lookup = ownersLookup(jobId); + const resolved = { owners: { buyer, seller }, buyerBundleRef: locator(951), sellerBundleRef: null }; + assert.deepEqual(await lookup(), resolved); + assert.deepEqual(await lookup(`&bundleRef=${locator(951)}`), resolved); + assert.deepEqual(await lookup(`&bundleRef=${locator(950)}`), { owners: null, buyerBundleRef: null }); + // So does a catalog entry the indexer could not verify, resolved through the same binding. + const catalog = store.loadCatalog(); + catalog.sellers[0].deals[0].refsVerified = false; + store.saveCatalog(catalog); + assert.deepEqual(await lookup(), resolved); + assert.deepEqual(await lookup(`&bundleRef=${locator(950)}`), { owners: null, buyerBundleRef: null }); + // The buyer's binding for the seller role does not bind the catalog's buyer ref. + const sellerScope = { ...scope, role: "seller", logicalAddress: logicalBundleAddress(jobId, "seller") }; + const sellerRole = { ...sellerScope, signature: { algorithm: "ed25519", signer: buyer, + value: await signature(contentHash(sellerScope), "dacs-bundle-binding:v1:", 0, "base64url") } } as BundleBinding; + assert.ok(await verifyBundleBinding(sellerRole)); + store.saveScanState({ ...store.loadScanState(), bundleBindings: { [jobId]: [sellerRole] } }); + assert.deepEqual(await lookup(`&bundleRef=${locator(951)}`), { owners: null, buyerBundleRef: null }); +}); + +test("a locator whose content was rejected stays periodic after later transient failures", () => { + resetIndex([later], 9); + const at = locator(500); + for (let pass = 0; pass < 4; pass++) { + store.recordArtifact({ locator: at, kind: "other", profile: "unknown", observedAt: Date.now(), rejected: true }); + store.recordArtifactFailure(at, "other", "ARTIFACT_REJECTED", "rejected", 1); + } + for (let attempt = 1; attempt <= 7; attempt++) { + store.recordArtifactFailure(at, "unknown", "STORAGE_RPC_UNAVAILABLE", "outage", 5); + const row = sql("SELECT status, next_retry_at FROM artifacts WHERE locator=?", at)[0]; + assert.equal(row.status, "retry", `transient failure ${attempt}`); + assert.ok(Number(row.next_retry_at) > 0); + } + assert.equal(sql("SELECT COUNT(*) count FROM dead_letters WHERE locator=?", at)[0].count, 0); + assert.deepEqual(sql("SELECT retry_count, rejection_count FROM artifacts WHERE locator=?", at)[0], { retry_count: 7, rejection_count: 4 }); + // A locator that only ever failed transiently keeps the bounded transient policy. + const transient = locator(501); + for (let attempt = 0; attempt < 5; attempt++) store.recordArtifactFailure(transient, "unknown", "STORAGE_RPC_UNAVAILABLE", "outage", 5); + assert.equal(sql("SELECT status FROM artifacts WHERE locator=?", transient)[0].status, "dead-letter"); +}); + +test("a pass reads at most 100 retries with transient failures first and at most 20 rejected", () => { + resetIndex([later], 16); + const transient = range(800, 120), rejected = range(950, 30); + for (const at of rejected) store.recordArtifactFailure(at, "other", "ARTIFACT_REJECTED", "rejected", 1); + for (const at of transient) store.recordArtifactFailure(at, "unknown", "STORAGE_RPC_UNAVAILABLE", "outage", 5); + const due = Date.now() + 4_000_000; + const full = store.loadRetryableArtifacts(due); + assert.deepEqual(full.slice(0, 80), transient.slice(0, 80)); + assert.equal(full.length, 100); + assert.ok(full.slice(80).every((at) => rejected.includes(at))); + // Transient retries fill whatever the rejected share does not use. + sql(`DELETE FROM artifacts WHERE locator IN (${rejected.slice(0, 25).map(() => "?").join(",")})`, ...rejected.slice(0, 25)); + const mixed = store.loadRetryableArtifacts(due); + assert.deepEqual(mixed.slice(0, 95), transient.slice(0, 95)); + assert.deepEqual(mixed.slice(95).sort(), rejected.slice(25).sort()); +}); + +test("rejected retries keep a reserved share while transient retries are due", () => { + resetIndex([later], 20); + const transient = range(8_000, 100), rejected = locator(8_500); + store.recordArtifactFailure(rejected, "other", "ARTIFACT_REJECTED", "rejected", 1); + for (const at of transient) store.recordArtifactFailure(at, "unknown", "STORAGE_RPC_UNAVAILABLE", "outage", 5); + assert.deepEqual(store.loadRetryableArtifacts(Date.now() + 4_000_000), [...transient.slice(0, 99), rejected]); +}); + +test("content that fails the artifact policy on a well-formed read is a rejection", async () => { + resetIndex([later], 21); + const named = locator(900), longName = locator(901); + chain({ [named]: { data: "[1,2]", name: "dacs5:bundle:bad" }, [longName]: { data: "{}", name: `dacs1:listing:${"x".repeat(1_100)}` } }, + [named, longName]); + await reindex(); + for (const at of [named, longName]) { + assert.deepEqual(sql("SELECT error_code, rejection_count, retry_count FROM artifacts WHERE locator=?", at)[0], + { error_code: "ARTIFACT_REJECTED", rejection_count: 1, retry_count: 0 }, at); + } + assert.equal(reads.get(named), 1, "content is not read again within the same pass"); +}); + +test("a successful read without content ends the rejected class", async () => { + resetIndex([later], 22); + const at = locator(910); + chain({ [at]: { data: '{"n":1e400}' } }, [at]); + await reindex(); + assert.equal(sql("SELECT rejection_count FROM artifacts WHERE locator=?", at)[0].rejection_count, 1); + transactions = []; + sql("UPDATE artifacts SET next_retry_at=0 WHERE locator=?", at); + await reindex(); + assert.equal(sql("SELECT rejection_count FROM artifacts WHERE locator=?", at)[0].rejection_count, 2, "rejections accumulate"); + storage[at] = { data: '"text"' }; + sql("UPDATE artifacts SET next_retry_at=0 WHERE locator=?", at); + await reindex(); + assert.deepEqual(sql("SELECT status, rejection_count, retry_count FROM artifacts WHERE locator=?", at)[0], + { status: "observed", rejection_count: 0, retry_count: 0 }); + // A later outage follows the bounded transient policy. + for (let attempt = 0; attempt < 5; attempt++) store.recordArtifactFailure(at, "unknown", "STORAGE_RPC_UNAVAILABLE", "outage", 5); + assert.equal(sql("SELECT status FROM artifacts WHERE locator=?", at)[0].status, "dead-letter"); +}); + +test("a transient retry is read before rejected retries and the revocation it carries is applied", async () => { + resetIndex([sellerRegistration, later], 10); + chain({ [locator(1)]: { data: JSON.stringify(listing) }, [locator(60)]: { data: JSON.stringify(marker) } }, [locator(1), locator(60)]); + const rejected = range(700, 150); + for (const at of rejected) { + store.recordArtifact({ locator: at, kind: "other", profile: "unknown", observedAt: Date.now(), rejected: true }); + store.recordArtifactFailure(at, "other", "ARTIFACT_REJECTED", "rejected", 1); + } + unavailable.add(locator(60)); + await reindex(); + assert.equal(listingStatus(), "active"); + // The marker's write is now outside the scan window; only its retry can read it again. + unavailable.clear(); + transactions = []; + reads = new Map(); + sql("UPDATE artifacts SET next_retry_at=0 WHERE error_code='ARTIFACT_REJECTED'"); + sql("UPDATE artifacts SET next_retry_at=1 WHERE locator=?", locator(60)); + await reindex(); + assert.equal(listingStatus(), "revoked"); + assert.ok(readsOf(rejected) <= 20, `rejected retries read ${readsOf(rejected)}`); +}); + +test("scheduled rejected retries have a ceiling and every waiting locator is still retried", () => { + resetIndex([later], 11); + const realNow = Date.now; + let clock = realNow(); + Date.now = () => clock; + try { + const total = 1_030; + const rejected = range(5_000, total); + const reject = (at: string) => { + store.recordArtifact({ locator: at, kind: "other", profile: "unknown", observedAt: clock, rejected: true }); + store.recordArtifactFailure(at, "other", "ARTIFACT_REJECTED", "rejected", 1); + }; + for (const at of rejected) reject(at); + const served = new Set(); + let rounds = 0; + while (served.size < total && rounds < 80) { + clock += 15 * 60_000; + rounds++; + const batch = store.loadRetryableArtifacts(); + assert.ok(batch.length <= 20, `round ${rounds} read ${batch.length} rejected locators`); + for (const at of batch) { served.add(at); reject(at); } + } + assert.equal(served.size, total, "no waiting locator is dropped"); + assert.ok(rounds <= Math.ceil(total / 20) + 2, `all served after ${rounds} rounds`); + const active = Number(sql("SELECT COUNT(*) count FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NULL")[0].count); + const waiting = Number(sql("SELECT COUNT(*) count FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NOT NULL")[0].count); + assert.ok(active <= 1_000, `${active} scheduled`); + assert.equal(active + waiting, total); + // Slots freed by recovered locators go to the longest-waiting ones on the next pass. + const recovered = (sql("SELECT locator FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NULL LIMIT 10")).map((row) => String(row.locator)); + for (const at of recovered) store.recordArtifact({ locator: at, kind: "other", profile: "unknown", contentHash: "e".repeat(64), observedAt: clock }); + store.loadRetryableArtifacts(); + assert.equal(Number(sql("SELECT COUNT(*) count FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NOT NULL")[0].count), waiting - 10); + } finally { Date.now = realNow; } +}); + +test("a stored registration over the binding limit is refused before its bindings are verified", async () => { + const scope = { bindingVersion: "1", jobId: "job", role: "buyer", logicalAddress: logicalBundleAddress("job", "buyer"), + nativeAddress: locator(10), bundleContentHash: "c".repeat(64), signer: buyer }; + const invalid = { ...scope, signature: { algorithm: "ed25519", signer: buyer, value: "A".repeat(86) } } as BundleBinding; + resetIndex([{ primaryClaim: buyer, displayName: "Over limit", listingAnchors: [], bundleBindings: Array(300).fill(invalid) }, later], 12); + chain({}, []); + await reindex(); + assert.deepEqual(store.loadCatalog().sellers.map((s) => s.displayName), ["Later"]); + assert.equal(lastRun().omitted_sellers, 1); + assert.equal(lastRun().omitted_bindings, 0, "no binding of a refused registration is verified"); +}); + +const resolverName = "dacs1:listing:bounded"; +const candidateSearch = () => Response.json({ result: 200, response: [{ storageAddress: locator(70), programName: resolverName }] }); +let cancelledStreams = 0; +/** A valid JSON body padded past the response ceiling. */ +function padded(head: string, tail: string) { + let sent = 0; + const encoder = new TextEncoder(); + return new Response(new ReadableStream({ + pull(controller) { + if (sent === 0) controller.enqueue(encoder.encode(head)); + const bytes = new Uint8Array(16_384).fill(32); + controller.enqueue(bytes); + sent += bytes.length; + if (sent > 1_100_000) { controller.enqueue(encoder.encode(tail)); controller.close(); } + }, + cancel() { cancelledStreams++; }, + })); +} +const resolverCases: Array<[string, typeof fetch, boolean]> = [ + ["search response", async () => padded('{"result":200,"response":[]', "}"), true], + ["candidate response", async (input) => String(input).includes("/storage-program/") + ? padded(`{"success":true,"owner":"${ownerOf(seller)}","programName":"${resolverName}","data":{"blob":"`, '"}}') + : candidateSearch(), true], + ["candidate artifact", async (input) => String(input).includes("/storage-program/") + ? new Response(JSON.stringify({ success: true, owner: ownerOf(seller), programName: resolverName, data: { blob: "x".repeat(300_000) } })) + : candidateSearch(), false], +]; +for (const [label, stub, streamed] of resolverCases) { + test(`the producer-side program resolver bounds an oversized ${label}`, async () => { + const original = globalThis.fetch; + cancelledStreams = 0; + globalThis.fetch = stub; + try { + assert.equal((await resolveOwnedAnchorByName(resolverName, ownerOf(seller))).status, "indeterminate"); + if (streamed) assert.equal(cancelledStreams, 1, "the response is cancelled rather than consumed"); + } finally { globalThis.fetch = original; } + }); +} + +const duplicateJob = "dup-job", duplicateName = `dacs3:agreement:${duplicateJob}`; +function duplicateProgramState() { + resetIndex([later], 13); + store.saveScanState({ ...emptyState(), programs: { [store.programBindingKey(ownerOf(buyer), duplicateName)]: null } }); + return deriveAnchorAddress(buyer, duplicateName); +} + +test("the artifact route reports an owner and name with duplicate programs as indeterminate", async () => { + const derived = duplicateProgramState(); + const response = await artifactRoute.GET(new NextRequest( + `http://localhost/api/dacs/artifact?owner=${ownerOf(buyer)}&name=${encodeURIComponent(duplicateName)}`)); + assert.equal(response.status, 409); + assert.notEqual((await response.json()).ref, derived); +}); + +test("the indexer never reads the derived address of an owner and name with duplicate programs", async () => { + const jobId = duplicateJob; + const derived = duplicateProgramState(); + const bundle = { bundleVersion: "0.1", jobId, outcome: "completed", anchoredByRole: "buyer", + listingRef: { listingId: "svc", version: 1, contentHash: listingHash }, + agreementRef: { kind: "dacs-3-agreement", id: "agreement", contentHash: "a".repeat(64) }, + parties: [{ role: "buyer", primaryClaim: buyer, bundleHash: "b" }, { role: "seller", primaryClaim: seller, bundleHash: "s" }], + phaseSummary: [], vetRecords: [], settlementEvidence: [], recipeRegistryVersion: 1, railRegistryVersion: 1, finalisedAt: 1 }; + chain({ [locator(80)]: { data: JSON.stringify(bundle), owner: ownerOf(buyer) } }, []); + const record = await indexRegistration({ primaryClaim: seller, displayName: "Seller", listingAnchors: [], + deals: [{ jobId, rail: "pay-dem", buyerBundleRef: locator(80), owners: { buyer, seller } }] }, undefined, offline); + assert.ok((reads.get(locator(80)) ?? 0) > 0, "the bundle itself was read"); + assert.equal(reads.get(derived) ?? 0, 0); + assert.equal(record.deals[0].refsVerified, false); +}); diff --git a/reference-implementations/dacs-directory/test/scan-robustness.test.ts b/reference-implementations/dacs-directory/test/scan-robustness.test.ts index 70e0f3c..be14e3a 100644 --- a/reference-implementations/dacs-directory/test/scan-robustness.test.ts +++ b/reference-implementations/dacs-directory/test/scan-robustness.test.ts @@ -29,6 +29,8 @@ const seeds = [21, 22].map((byte) => Uint8Array.from(Buffer.alloc(32, byte))); const dids = seeds.map((seed) => `did:demos:agent:${Buffer.from(rawPublicKey(publicKeyFromSeed(seed))).toString("hex")}`); const [buyer, seller] = dids; const ownerOf = (did: string) => `0x${did.slice(-64)}`; +/** Discovered deals are keyed by their buyer and jobId. */ +const dealKey = (jobId: string) => `${ownerOf(buyer)}\n${jobId}`; const outsider = `did:demos:agent:${"d".repeat(64)}`; const locator = (n: number) => `stor-${n.toString(16).padStart(40, "0")}`; const offline = async (): Promise => { throw new Error("identity unavailable"); }; @@ -152,12 +154,12 @@ for (const jobId of ["constructor", "__proto__", "toString", "hasOwnProperty"]) chain({ [locator(1)]: { name: `dacs5:bundle:${jobId}`, data: '{"any":1}', owner: ownerOf(buyer) }, [locator(2)]: { name: `dacs3:agreement:${jobId}`, data: JSON.stringify({ seller }), owner: ownerOf(buyer) } }, [locator(1), locator(2)]); await reindex(); - assert.ok(Object.hasOwn(store.loadScanState().deals, jobId)); - assert.equal(store.loadScanState().deals[jobId].owners.seller, ""); + assert.ok(Object.hasOwn(store.loadScanState().deals, dealKey(jobId))); + assert.equal(store.loadScanState().deals[dealKey(jobId)].owners.seller, ""); transactions = []; await reindex(); assert.ok(store.loadCatalog().generatedAt > 5); - assert.ok(Object.hasOwn(store.loadScanState().deals, jobId)); + assert.ok(Object.hasOwn(store.loadScanState().deals, dealKey(jobId))); assert.ok(store.loadCatalog().sellers.some((s) => s.displayName === "Later")); }); test(`signed reserved binding ${jobId} survives two reindexes and SQLite reload`, async () => { @@ -286,19 +288,19 @@ test("rejected revocation outside replay overlap is retried and consensus time r test("rejected retries remain periodic after exhaustion with a bounded queue", () => { sql("DELETE FROM artifacts RETURNING locator"); for (let i = 0; i < 120; i++) for (let pass = 0; pass < 8; pass++) { - store.recordArtifact({ locator: locator(1000 + i), kind: "other", profile: "unknown", observedAt: Date.now() }); + store.recordArtifact({ locator: locator(1000 + i), kind: "other", profile: "unknown", observedAt: Date.now(), rejected: true }); store.recordArtifactFailure(locator(1000 + i), "other", "ARTIFACT_REJECTED", "rejected", 1); } - const retry = sql("SELECT retry_count,status,next_retry_at FROM artifacts WHERE locator='" + locator(1000) + "'")[0]; - assert.equal(retry.retry_count, 8); + const retry = sql("SELECT rejection_count,status,next_retry_at FROM artifacts WHERE locator='" + locator(1000) + "'")[0]; + assert.equal(retry.rejection_count, 8); assert.equal(retry.status, "retry"); assert.ok(Number(retry.next_retry_at) > Date.now(), "periodic rejection retries have a future deadline"); assert.ok(Number(retry.next_retry_at) <= Date.now() + 3_600_000, "backoff is capped at one hour"); // Pre-fix ARTIFACT_REJECTED dead letters had no retry deadline. sql("UPDATE artifacts SET status='dead-letter', next_retry_at=NULL WHERE locator='" + locator(1000) + "' RETURNING locator"); const queued = store.loadRetryableArtifacts(Date.now() + 4_000_000); - assert.equal(queued.length, 100); - assert.equal(new Set(queued).size, 100); + assert.equal(queued.length, 20, "rejected retries have their own per-pass budget"); + assert.equal(new Set(queued).size, 20); assert.ok(queued.includes(locator(1000)), "old rejected dead letters are due for another read"); }); test("duplicate programs are indeterminate within a pass and across passes", async () => { @@ -312,19 +314,19 @@ test("duplicate programs are indeterminate within a pass and across passes", asy chain({ [locator(20)]: bundle, [locator(21)]: honest, [locator(22)]: invalid }, order); const result = await scan(); assert.equal(result.programs.get(key), null); - assert.equal(result.deals.get(jobId)?.owners.seller, ""); + assert.equal(result.deals.get(dealKey(jobId))?.owners.seller, ""); } resetIndex([later], 10); chain({ [locator(20)]: bundle, [locator(21)]: honest }, [locator(20), locator(21)]); await reindex(); - assert.equal(store.loadScanState().deals[jobId].owners.seller, seller); + assert.equal(store.loadScanState().deals[dealKey(jobId)].owners.seller, seller); storage[locator(22)] = invalid; transactions = [{ id: 100, content: JSON.stringify({ memo: locator(22) }) }]; await reindex(); assert.equal(store.findProgramAddress(ownerOf(buyer), name), null); - assert.equal(store.loadScanState().deals[jobId].owners.seller, ""); + assert.equal(store.loadScanState().deals[dealKey(jobId)].owners.seller, ""); await reindex(); - assert.equal(store.loadScanState().deals[jobId].owners.seller, ""); + assert.equal(store.loadScanState().deals[dealKey(jobId)].owners.seller, ""); }); test("unrelated non-object storage stays unclassified while DACS-named data is rejected", async () => { chain({ [locator(30)]: { data: '"text"' }, [locator(31)]: { data: '[1,2]' }, [locator(32)]: { data: '[]', name: 'dacs5:bundle:bad' } }, [locator(30), locator(31), locator(32)]); diff --git a/reference-implementations/dacs-directory/test/startup-migration.test.ts b/reference-implementations/dacs-directory/test/startup-migration.test.ts new file mode 100644 index 0000000..513c0a8 --- /dev/null +++ b/reference-implementations/dacs-directory/test/startup-migration.test.ts @@ -0,0 +1,79 @@ +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"; + +// A database written by the previous release: separate retry classes, but more +// scheduled rejected retries than the ceiling, and a revocation candidate queue +// carried in scan-state JSON. Startup migrates both in place. +const dataDirectory = mkdtempSync(join(tmpdir(), "dacs-directory-startup-")); +const locator = (n: number) => `stor-${n.toString(16).padStart(40, "0")}`; +const rejected = Array.from({ length: 1_030 }, (_, index) => locator(1 + index)); +const listingHash = "a".repeat(64); +const listingOwner = `0x${"B".repeat(64)}`; +const [other, owned, unknown] = [locator(5_001), locator(5_002), locator(5_003)]; +const seeded = new Database(join(dataDirectory, "directory.sqlite")); +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), ('sr2-anchor-schema-version','2',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, + rejection_count INTEGER NOT NULL DEFAULT 0, deferred_at INTEGER + ); +`); +const insert = seeded.prepare(`INSERT INTO artifacts(locator,kind,profile,observed_at,status,error_code,error_message,next_retry_at,rejection_count) + VALUES (?,'other','unknown',1,'retry','ARTIFACT_REJECTED','rejected',?,1)`); +// The first 1,000 are due in locator order; the 30 beyond them are due in reverse locator order. +for (const [index, at] of rejected.entries()) insert.run(at, index < 1_000 ? 1_000 + index : 4_029 - index); +const observed = seeded.prepare("INSERT INTO artifacts(locator,kind,profile,owner,content_hash,observed_at) VALUES (?,'listing-revocation','dacs-v0.1',?,?,1)"); +observed.run(other, `0x${"c".repeat(64)}`, "1".repeat(64)); +observed.run(owned, listingOwner, "2".repeat(64)); +seeded.prepare("INSERT INTO kv_state(key,value_json,updated_at) VALUES ('scan-state',?,0)").run(JSON.stringify({ + schemaVersion: 9, lastSeenTxId: 7, listings: {}, deals: {}, programs: {}, + revocations: { [listingHash]: [other, owned, unknown] }, verifiedRevocations: {}, +})); +seeded.close(); + +process.env.DACS_DIRECTORY_DATA = dataDirectory; +const store = await import("../src/catalog/store.js"); + +test.after(() => rmSync(dataDirectory, { recursive: true, force: true })); + +test("startup defers scheduled rejected retries beyond the ceiling, latest due first", () => { + const db = new Database(join(dataDirectory, "directory.sqlite"), { readonly: true }); + try { + const scheduled = db.prepare("SELECT locator FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NULL ORDER BY next_retry_at") + .all() as Array<{ locator: string }>; + assert.deepEqual(scheduled.map((row) => row.locator), rejected.slice(0, store.MAX_ACTIVE_REJECTED_RETRIES)); + const waiting = db.prepare("SELECT locator FROM artifacts WHERE deferred_at IS NOT NULL AND next_retry_at IS NULL ORDER BY locator") + .all() as Array<{ locator: string }>; + assert.deepEqual(waiting.map((row) => row.locator), rejected.slice(store.MAX_ACTIVE_REJECTED_RETRIES)); + } finally { db.close(); } + assert.equal(store.loadRetryableArtifacts(10_000).length, store.REJECTED_RETRY_READS_PER_PASS); +}); + +test("rejected retries deferred at startup return to the schedule in due order", () => { + // Five scheduled locators read successfully, which frees five slots. + for (const at of rejected.slice(0, 5)) store.recordArtifact({ locator: at, kind: "other", profile: "unknown", observedAt: 2 }); + store.loadRetryableArtifacts(10_000); + const db = new Database(join(dataDirectory, "directory.sqlite"), { readonly: true }); + try { + const returned = db.prepare(`SELECT locator FROM artifacts WHERE rejection_count > 0 AND deferred_at IS NULL + AND locator IN (${rejected.slice(store.MAX_ACTIVE_REJECTED_RETRIES).map(() => "?").join(",")}) ORDER BY locator`) + .all(...rejected.slice(store.MAX_ACTIVE_REJECTED_RETRIES)) as Array<{ locator: string }>; + assert.deepEqual(returned.map((row) => row.locator), rejected.slice(-5)); + } finally { db.close(); } +}); + +test("startup moves a candidate queue from scan state into the candidate store in order", () => { + assert.deepEqual(Object.keys(store.loadScanState().revocations), []); + assert.equal(store.loadScanState().lastSeenTxId, 7); + // Arrival order is kept, and candidates anchored by the listing's owner come first. + assert.deepEqual(store.nextRevocationCandidates(listingHash, null, 16), [other, owned, unknown]); + assert.deepEqual(store.nextRevocationCandidates(listingHash, `did:demos:agent:${"b".repeat(64)}`, 16), [owned, other, unknown]); + assert.deepEqual(store.nextRevocationCandidates(listingHash, null, 2), [other, owned]); +}); diff --git a/reference-implementations/dacs-directory/test/verification.test.ts b/reference-implementations/dacs-directory/test/verification.test.ts index 6c7857c..fef7b72 100644 --- a/reference-implementations/dacs-directory/test/verification.test.ts +++ b/reference-implementations/dacs-directory/test/verification.test.ts @@ -30,11 +30,7 @@ import { verifyListing, verifyListingRevocation, } from "../src/catalog/listingVerification.js"; -import { - addRevocationCandidate, - isListingRevocationCandidate, - MAX_REVOCATION_CANDIDATES_PER_LISTING, -} from "../src/catalog/scan.js"; +import { addRevocationCandidate, isListingRevocationCandidate } from "../src/catalog/scan.js"; import { deriveSellerReputation, flipOutcome } from "../src/catalog/reputation.js"; import type { Catalog, DealRecord, SellerRecord } from "../src/catalog/types.js"; import type { BundleVerification } from "../vendor/dacs-sdk/dist/agent/verifyBundleCore.js"; @@ -581,11 +577,11 @@ test("any valid revocation candidate wins and scanner candidates deduplicate", a "dacs1-revoked:cci-xm%3Aevm%3Amainnet%3A0x1234:offer:v3", ); - const candidates = new Map(); + const candidates = new Map>(); addRevocationCandidate(candidates, verified.contentHash, "stor-old"); addRevocationCandidate(candidates, verified.contentHash, "stor-new"); addRevocationCandidate(candidates, verified.contentHash, "stor-old"); - assert.deepEqual(candidates.get(verified.contentHash), ["stor-old", "stor-new"]); + assert.deepEqual([...candidates.get(verified.contentHash) ?? []], ["stor-old", "stor-new"]); // Discovery is value-based: a conforming producer may use any colon-free, // opaque StorageProgram name and the marker remains discoverable. @@ -700,41 +696,43 @@ test("a revocation candidate with no canonical form never stops the candidate lo assert.equal(await hasValidListingRevocation(["infinite", "fractional"], verified, 1, read), false); }); -test("revocation candidate discovery is bounded per listing hash", () => { - const candidates = new Map(); +test("revocation candidate discovery admits every locator once", () => { + const candidates = new Map>(); const listingHash = "a".repeat(64); - for (let index = 0; index < 20; index++) { + for (let index = 0; index < 40; index++) { addRevocationCandidate(candidates, listingHash, `stor-${index.toString(16).padStart(40, "0")}`); } - assert.equal(candidates.get(listingHash)?.length, MAX_REVOCATION_CANDIDATES_PER_LISTING); + addRevocationCandidate(candidates, listingHash, `stor-${"0".repeat(40)}`); + const admitted = [...candidates.get(listingHash) ?? []]; + assert.equal(admitted.length, 40); + assert.equal(admitted[0], `stor-${"0".repeat(40)}`); }); -test("candidate pruning preserves an RB-4-verified locator across restart", () => { - const listingHash = "b".repeat(64); - const verified = `stor-${"f".repeat(40)}`; - const beforeRestart = new Map(); - for (let index = 0; index < MAX_REVOCATION_CANDIDATES_PER_LISTING; index++) { - addRevocationCandidate(beforeRestart, listingHash, `stor-${index.toString(16).padStart(40, "0")}`); - } - addRevocationCandidate(beforeRestart, listingHash, verified, new Set([verified])); - - const persisted = JSON.parse(JSON.stringify(Object.fromEntries(beforeRestart))) as Record; - const afterRestart = new Map(Object.entries(persisted)); - for (let index = 16; index < 32; index++) { - addRevocationCandidate( - afterRestart, - listingHash, - `stor-${index.toString(16).padStart(40, "0")}`, - new Set([verified]), - ); - } - - assert.equal(afterRestart.get(listingHash)?.length, MAX_REVOCATION_CANDIDATES_PER_LISTING); - assert.equal(afterRestart.get(listingHash)?.includes(verified), true); - - const staleState = new Map([[listingHash, afterRestart.get(listingHash)!.filter((address) => address !== verified)]]); - addRevocationCandidate(staleState, listingHash, `stor-${"e".repeat(40)}`, new Set([verified])); - assert.equal(staleState.get(listingHash)?.includes(verified), true); +test("revocation verification work per call is bounded and reports what it examined", async () => { + const perPass = 16; + const listingMessage = Buffer.from(`dacs-listing:v1:${contentHash(listing)}`, "utf8"); + const listingSignature = Buffer.from(await ed25519Sign(listingMessage, privateKeyFromSeed(seed))).toString("hex"); + const verified = await verifyListing({ ...listing, signature: { algorithm: "ed25519", signer: did, value: listingSignature } }); + assert.ok(verified); + if (!verified) return; + const scope = { listingId: listing.listingId, listingVersion: listing.listingVersion, listingContentHash: verified.contentHash, revokedAt: 1 }; + const value = Buffer.from(await ed25519Sign(Buffer.from(`dacs-revocation:v1:${contentHash(scope)}`, "utf8"), privateKeyFromSeed(seed))).toString("hex"); + const valid = { ...scope, signature: { algorithm: "ed25519", signer: did, value } }; + const refs = Array.from({ length: 20 }, (_, index) => `bogus-${index}`); + const reads: string[] = []; + const read = async (ref: string) => { + reads.push(ref); + if (ref === "valid") return valid; + return ref === "bogus-3" ? null : { ...valid, revokedAt: 2 }; + }; + const progress = { unread: [] as string[], rejected: [] as string[] }; + assert.equal(await findValidListingRevocation([...refs, "valid"], verified, 1, read, undefined, progress), null); + assert.equal(reads.length, perPass); + assert.deepEqual(progress.unread, ["bogus-3"]); + assert.equal(progress.rejected.length, perPass - 1); + const remaining = [...refs, "valid"].filter((ref) => !progress.rejected.includes(ref) && !progress.unread.includes(ref)); + const binding = await findValidListingRevocation([...remaining, ...progress.unread], verified, 1, read); + assert.equal(binding?.markerAnchor.locator, "valid"); }); test("public discovery excludes revoked listings and empty sellers", () => {