Per-order terminal reorg linkage
This commit is contained in:
@@ -11,7 +11,8 @@ const pageSchema = z.object({ streamId: z.string().uuid(), previousAnchor: hex,
|
||||
events: z.array(eventSchema).max(64), cursor: z.number().int().nonnegative().safe(),
|
||||
head: z.number().int().nonnegative().safe(), caughtUp: z.boolean() }).strict();
|
||||
export type CanonicalInboxPage = z.infer<typeof pageSchema>;
|
||||
export type CanonicalInboxResult = { streamId: string; cursor: number; head: number; height: number; tip: string; replayed: boolean };
|
||||
export type CanonicalInboxResult = { streamId: string; cursor: number; head: number; height: number; tip: string;
|
||||
replayed: boolean; appendedEvents: z.infer<typeof eventSchema>[] };
|
||||
|
||||
export class CanonicalInboxError extends Error {
|
||||
readonly code: string;
|
||||
@@ -96,7 +97,7 @@ export async function ingestCanonicalPage(client: PoolClient, raw: unknown): Pro
|
||||
const savedAnchor = await anchorAt(client, page.cursor);
|
||||
if (page.cursor > cursor || savedAnchor !== page.anchor || page.previousAnchor !== page.anchor) throw new CanonicalInboxError('empty-cursor-mismatch');
|
||||
return { streamId: page.streamId, cursor, head: Math.max(page.head, cursor), height: stack.length,
|
||||
tip: stack.at(-1) ?? GENESIS_HASH, replayed: true };
|
||||
tip: stack.at(-1) ?? GENESIS_HASH, replayed: true, appendedEvents: [] };
|
||||
}
|
||||
const firstSequence = page.events[0].sequence;
|
||||
if (firstSequence > cursor + 1) throw new CanonicalInboxError('cursor-gap');
|
||||
@@ -104,6 +105,7 @@ export async function ingestCanonicalPage(client: PoolClient, raw: unknown): Pro
|
||||
if (!predecessor || page.previousAnchor !== predecessor) throw new CanonicalInboxError('cursor-anchor-mismatch');
|
||||
let priorAnchor = predecessor;
|
||||
let appended = false;
|
||||
const appendedEvents: z.infer<typeof eventSchema>[] = [];
|
||||
for (const e of page.events) {
|
||||
if (e.eventHash !== eventDigest(page.streamId, e, priorAnchor)) throw new CanonicalInboxError('event-hash-invalid');
|
||||
if (e.sequence <= cursor) {
|
||||
@@ -113,13 +115,13 @@ export async function ingestCanonicalPage(client: PoolClient, raw: unknown): Pro
|
||||
if (e.sequence !== cursor + 1 || priorAnchor !== anchor) throw new CanonicalInboxError('cursor-anchor-mismatch');
|
||||
applyTip(stack, e);
|
||||
await client.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)', [eventKey(e.sequence), JSON.stringify(e)]);
|
||||
cursor = e.sequence; anchor = e.eventHash; appended = true;
|
||||
cursor = e.sequence; anchor = e.eventHash; appended = true; appendedEvents.push(e);
|
||||
}
|
||||
priorAnchor = e.eventHash;
|
||||
}
|
||||
if (page.cursor !== page.events.at(-1)!.sequence || page.anchor !== priorAnchor) throw new CanonicalInboxError('cursor-anchor-mismatch');
|
||||
return { streamId: page.streamId, cursor, head: Math.max(page.head, cursor), height: stack.length,
|
||||
tip: stack.at(-1) ?? GENESIS_HASH, replayed: !appended };
|
||||
tip: stack.at(-1) ?? GENESIS_HASH, replayed: !appended, appendedEvents };
|
||||
}
|
||||
|
||||
/** Full append-only inbox verification. Expected to run under the caller's coherent audit snapshot. */
|
||||
|
||||
@@ -98,7 +98,8 @@ async function checkOrders(client: PoolClient, rows: OrderRow[]) {
|
||||
c(`customer:${o.seller}:available`, o.sellerAtomicXmr), c(`author:${o.authorAddress}`, o.authorAtomicXmr),
|
||||
c(o.hostFeeAddress ? `revenue:host:${o.hostFeeAddress}` : 'revenue:host', o.hostAtomicXmr)] : null);
|
||||
const evidenceRows = await client.query('SELECT key,value FROM wk_custody.custody_meta WHERE key=ANY($1::text[])',
|
||||
[[`sale-rehearsal-binding:${o.id}`, `sale-rehearsal-observation:${o.id}`, `sale-rehearsal-terminal:${o.id}`]]);
|
||||
[[`sale-rehearsal-binding:${o.id}`, `sale-rehearsal-observation:${o.id}`, `sale-rehearsal-terminal:${o.id}`,
|
||||
`sale-rehearsal-terminal-index:${o.id}`]]);
|
||||
if (evidenceRows.rowCount) {
|
||||
try {
|
||||
const records = new Map<string, string>(evidenceRows.rows.map(r => [r.key, r.value]));
|
||||
@@ -113,11 +114,24 @@ async function checkOrders(client: PoolClient, rows: OrderRow[]) {
|
||||
};
|
||||
verify(`sale-rehearsal-observation:${o.id}`);
|
||||
if (o.status === 'awaiting-lock') {
|
||||
if (records.has(`sale-rehearsal-terminal:${o.id}`)) fail('sale-rehearsal-evidence-mismatch');
|
||||
if (records.has(`sale-rehearsal-terminal:${o.id}`) || records.has(`sale-rehearsal-terminal-index:${o.id}`)) {
|
||||
fail('sale-rehearsal-evidence-mismatch');
|
||||
}
|
||||
} else {
|
||||
const replay = verify(`sale-rehearsal-terminal:${o.id}`);
|
||||
if (replay.decision.action !== (o.status === 'settled' ? 'settle' : 'refund') ||
|
||||
(o.status === 'settled' && row.box_data !== `fixture:${replay.stateRoot}`)) fail('sale-rehearsal-evidence-mismatch');
|
||||
const index = JSON.parse(records.get(`sale-rehearsal-terminal-index:${o.id}`) ?? '');
|
||||
if (index.version !== 1 || index.stateRoot !== replay.stateRoot ||
|
||||
index.terminalTransactionId !== (replay.terminalTransactionId ?? null)) fail('sale-rehearsal-evidence-mismatch');
|
||||
if (replay.terminalTransactionId) {
|
||||
const link = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1',
|
||||
[`sale-rehearsal-tx:${replay.terminalTransactionId}`]);
|
||||
const parsedLink = JSON.parse(link.rows[0]?.value ?? '');
|
||||
if (parsedLink.version !== 1 || parsedLink.orderId !== o.id || parsedLink.stateRoot !== replay.stateRoot) {
|
||||
fail('sale-rehearsal-terminal-index-mismatch');
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch { fail('sale-rehearsal-evidence-mismatch'); }
|
||||
}
|
||||
|
||||
@@ -334,6 +334,14 @@ export class PostgresCustody {
|
||||
await this.update(c, 'orders', o); await this.update(c, 'listings', l);
|
||||
await c.query('INSERT INTO wk_custody.custody_meta VALUES($1,$2)', [`sale-rehearsal-terminal:${o.id}`,
|
||||
JSON.stringify({ version: 1, binding: replay.binding, stateRoot: replay.stateRoot, input: replay.input })]);
|
||||
await c.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)',
|
||||
[`sale-rehearsal-terminal-index:${o.id}`, JSON.stringify({ version: 1,
|
||||
terminalTransactionId: replay.terminalTransactionId ?? null, stateRoot: replay.stateRoot })]);
|
||||
if (replay.terminalTransactionId) {
|
||||
await c.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)',
|
||||
[`sale-rehearsal-tx:${replay.terminalTransactionId}`,
|
||||
JSON.stringify({ version: 1, orderId: o.id, stateRoot: replay.stateRoot })]);
|
||||
}
|
||||
await auditOrder(c, o.id);
|
||||
}
|
||||
return { order: o, action: replay.decision.action, reservationDigest: replay.reservationDigest, stateRoot: replay.stateRoot };
|
||||
@@ -345,7 +353,17 @@ export class PostgresCustody {
|
||||
return this.transaction(true, async c => {
|
||||
check(!this.diagnostics().mutationHold, 'Custody reconciliation hold');
|
||||
check(!(await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'")).rowCount, 'Custody integrity incident unresolved');
|
||||
try { return await ingestCanonicalPage(c, page); }
|
||||
try {
|
||||
const result = await ingestCanonicalPage(c, page);
|
||||
for (const event of result.appendedEvents) {
|
||||
if (event.type !== 'disconnect') continue;
|
||||
for (const txid of event.transactionIds) {
|
||||
const terminal = await c.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [`sale-rehearsal-tx:${txid}`]);
|
||||
if (terminal.rowCount) throw new CustodyIntegrityError('posted-sale-terminal-disconnected');
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
catch (error) {
|
||||
if (error instanceof CanonicalInboxError) throw new CustodyIntegrityError(`canonical-inbox-${error.code}`);
|
||||
throw error;
|
||||
|
||||
@@ -3,7 +3,7 @@ import { ADDITIONAL_MARKET_FEE } from '../../shared/market.ts';
|
||||
import type { CustodyOrder } from '../../shared/custody.ts';
|
||||
import { saleTransactionSchema, saleTermsMessage } from '../../shared/sale-candidate.ts';
|
||||
import { applySaleBlock, initialSaleState, saleContext, saleDecision, saleId, saleStateRoot,
|
||||
termsForCustodyOrder, type SaleContext } from '../testnet/sale-candidate.ts';
|
||||
saleCandidateTransactionId, termsForCustodyOrder, type SaleContext } from '../testnet/sale-candidate.ts';
|
||||
import { sha256, verifySignature } from '../testnet/consensus.ts';
|
||||
|
||||
const height = z.number().int().min(0).max(1000000);
|
||||
@@ -71,14 +71,17 @@ export function replayReservedSale(order: CustodyOrder, custodianId: string, raw
|
||||
order.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address, binding]));
|
||||
let state = initialSaleState({ [order.cardHash]: { owner: order.seller, sequence: order.sequence,
|
||||
mintedHeight: input.seed.mintedHeight, pageId: input.seed.pageId } }, input.seed.height);
|
||||
let lockTransactionId: string | undefined, terminalTransactionId: string | undefined;
|
||||
for (const block of input.blocks) {
|
||||
for (const tx of block) {
|
||||
if (tx.type === 'sale-lock' ? saleId(tx.terms) !== saleId(terms) : tx.saleId !== saleId(terms)) throw new Error('Unrelated sale in rehearsal');
|
||||
if (tx.type === 'sale-complete' && tx.reservationDigest !== reservationDigest) throw new Error('Completion reservation commitment mismatch');
|
||||
if (tx.type === 'sale-lock') lockTransactionId = saleCandidateTransactionId(tx);
|
||||
else terminalTransactionId = saleCandidateTransactionId(tx);
|
||||
}
|
||||
state = applySaleBlock(state, block, context);
|
||||
}
|
||||
const previous = order.status === 'settled' ? 'settled' : order.status === 'cancelled' ? 'refunded' : 'reserved';
|
||||
return { input, binding, reservationDigest, stateRoot: saleStateRoot(state, context),
|
||||
return { input, binding, reservationDigest, lockTransactionId, terminalTransactionId, stateRoot: saleStateRoot(state, context),
|
||||
decision: saleDecision(state, terms, context, previous), context };
|
||||
}
|
||||
|
||||
@@ -128,11 +128,19 @@ export function applySaleBlock(parent: SaleState, raw: unknown[], context: SaleC
|
||||
}
|
||||
return state;
|
||||
}
|
||||
/** Candidate transaction IDs exclude malleable signature bytes, matching the active chain convention.
|
||||
* These IDs are an unactivated candidate format and are NOT accepted by the existing Ledger schema.
|
||||
*/
|
||||
export function saleCandidateTransactionId(tx: SaleTransaction): string {
|
||||
if (tx.type === 'sale-lock') return sha256(JSON.stringify([SALE_CANDIDATE, 'transaction', 'lock', saleTermsMessage(tx.terms)]));
|
||||
if (tx.type === 'sale-complete') return sha256(JSON.stringify([SALE_CANDIDATE, 'transaction', 'complete', saleCompletionMessage(tx)]));
|
||||
if (tx.type === 'sale-cancel') return sha256(JSON.stringify([SALE_CANDIDATE, 'transaction', 'cancel', saleCancellationMessage(tx)]));
|
||||
return sha256(JSON.stringify([SALE_CANDIDATE, 'transaction', 'expire', tx.chainId, tx.saleId]));
|
||||
}
|
||||
function candidateIdentity(tx: CandidateTransaction): string {
|
||||
if (tx.type === 'sale-lock') return `lock:${saleId(tx.terms)}`;
|
||||
if (tx.type === 'sale-complete') return sha256(saleCompletionMessage(tx));
|
||||
if (tx.type === 'sale-cancel') return sha256(saleCancellationMessage(tx));
|
||||
if (tx.type === 'sale-expire') return `expire:${tx.chainId}:${tx.saleId}`;
|
||||
if (tx.type === 'sale-lock' || tx.type === 'sale-complete' || tx.type === 'sale-cancel' || tx.type === 'sale-expire') {
|
||||
return saleCandidateTransactionId(tx);
|
||||
}
|
||||
return sha256(tx.type === 'native-transfer' ? nftTransferMessage(tx.transfer) : nftTradeMessage(tx.trade));
|
||||
}
|
||||
function applyNative(state: SaleState, tx: Extract<CandidateTransaction, { type: 'native-transfer' | 'native-trade' }>, context: SaleContext) {
|
||||
|
||||
Reference in New Issue
Block a user