feat: publish Vector proof anchors per Base settlement
This commit is contained in:
david
@@ -160,3 +160,18 @@ SURPLUS_REQUEST_TIMEOUT_MS=120000
|
||||
SURPLUS_STOP_AFTER_SETTLED_REQUESTS=0
|
||||
SURPLUS_SETTLED_STATUSES=confirmed
|
||||
SURPLUS_STATE_PATH=/var/lib/surplus-manager/state.json
|
||||
# --- Vector proof publisher ------------------------------------------------
|
||||
# When enabled, every settled Base settlement transaction gets one Vector
|
||||
# counterpart: the sales it settled are Merkle-batched and anchored in a
|
||||
# label-674 metadata transaction signed by the proof wallet below.
|
||||
SURPLUS_PROOF_ENABLED=
|
||||
# Required when enabled. Mainnet Ogmios endpoint and the 64-hex Ed25519 key
|
||||
# of the AP3X proof wallet (the retired Vector supplier wallet).
|
||||
OGMIOS_URL=
|
||||
SURPLUS_PROOF_WALLET_PRIV_KEY_HEX=
|
||||
SURPLUS_PROOF_LEDGER_PATH=/var/lib/surplus-manager/vector-proofs.json
|
||||
# Anchoring pauses when the wallet cannot keep this reserve after one fee.
|
||||
SURPLUS_PROOF_RESERVE_LOVELACE=5000000
|
||||
SURPLUS_PROOF_FEE_BUDGET_LOVELACE=1000000
|
||||
SURPLUS_PROOF_INTERVAL_MS=300000
|
||||
SURPLUS_PROOF_CONFIRM_TIMEOUT_MS=300000
|
||||
|
||||
@@ -15,10 +15,20 @@ import {
|
||||
SurplusDashboardService,
|
||||
type SurplusDashboardSnapshot,
|
||||
} from "../surplus/dashboard.js";
|
||||
import { FileSurplusVectorProofLedger } from "../surplus/vectorProof.js";
|
||||
import { SurplusVectorProofPublisher } from "../surplus/vectorPublisher.js";
|
||||
import { LiveOgmiosProvider } from "@marketplace/shared/chain";
|
||||
import type { WalletKey } from "@marketplace/shared/tx";
|
||||
import { buildAnchorMetadataTx } from "@marketplace/shared/tx/server";
|
||||
import { deriveKeypair } from "./gen-keypair.js";
|
||||
|
||||
export async function runSurplusSeller(
|
||||
env: Record<string, string | undefined>,
|
||||
): Promise<{ server: Server; controller: SurplusSellerController }> {
|
||||
): Promise<{
|
||||
server: Server;
|
||||
controller: SurplusSellerController;
|
||||
publisher: SurplusVectorProofPublisher | null;
|
||||
}> {
|
||||
const config = loadSurplusManagerConfig(env);
|
||||
const client = new SurplusClient({
|
||||
apiBaseUrl: config.apiBaseUrl,
|
||||
@@ -38,11 +48,51 @@ export async function runSurplusSeller(
|
||||
stateStore,
|
||||
log: (message) => console.log(message),
|
||||
});
|
||||
|
||||
let publisher: SurplusVectorProofPublisher | null = null;
|
||||
let proofs: FileSurplusVectorProofLedger | undefined;
|
||||
if (config.proof !== null) {
|
||||
const proofConfig = config.proof;
|
||||
const ledger = await FileSurplusVectorProofLedger.open(proofConfig.ledgerPath);
|
||||
proofs = ledger;
|
||||
const chain = new LiveOgmiosProvider({ ogmiosUrl: proofConfig.ogmiosUrl });
|
||||
const derived = deriveKeypair(proofConfig.walletPrivateKeyHex, 1);
|
||||
const walletKey: WalletKey = {
|
||||
pubKeyHash: derived.pubKeyHash,
|
||||
pubKeyHex: derived.publicKeyHex,
|
||||
privateKeyHex: derived.privateKeyHex,
|
||||
address: derived.address,
|
||||
};
|
||||
publisher = new SurplusVectorProofPublisher({
|
||||
ledger,
|
||||
earnings: () => client.getEarnings(),
|
||||
anchor: async (metadata) => {
|
||||
const built = await buildAnchorMetadataTx({ chain, walletKey, metadata });
|
||||
return { expectedTxHash: built.expectedTxHash };
|
||||
},
|
||||
awaitTx: (txHash, timeoutMs) => chain.awaitTx(txHash, timeoutMs),
|
||||
balanceLovelace: async () => {
|
||||
const utxos = await chain.queryUtxosByAddress(walletKey.address);
|
||||
return utxos.reduce((sum, utxo) => sum + utxo.lovelace, 0n);
|
||||
},
|
||||
reserveLovelace: proofConfig.reserveLovelace,
|
||||
feeBudgetLovelace: proofConfig.feeBudgetLovelace,
|
||||
settledStatuses: config.settledStatuses,
|
||||
intervalMs: proofConfig.intervalMs,
|
||||
confirmTimeoutMs: proofConfig.confirmTimeoutMs,
|
||||
log: (message) => console.log(message),
|
||||
});
|
||||
console.log(
|
||||
`surplus-proof: publisher enabled with wallet ${derived.address} and ledger ${proofConfig.ledgerPath}`,
|
||||
);
|
||||
}
|
||||
|
||||
const dashboard = new SurplusDashboardService({
|
||||
controller,
|
||||
client,
|
||||
allowance,
|
||||
config,
|
||||
proofs,
|
||||
});
|
||||
const server = createSurplusServer(controller, dashboard);
|
||||
await new Promise<void>((resolveListen, rejectListen) => {
|
||||
@@ -53,7 +103,8 @@ export async function runSurplusSeller(
|
||||
});
|
||||
});
|
||||
await controller.start();
|
||||
return { server, controller };
|
||||
publisher?.start();
|
||||
return { server, controller, publisher };
|
||||
}
|
||||
|
||||
interface DashboardSource {
|
||||
@@ -115,13 +166,14 @@ async function handleRequest(
|
||||
export async function main(
|
||||
env: Record<string, string | undefined> = process.env,
|
||||
): Promise<void> {
|
||||
const { server, controller } = await runSurplusSeller(env);
|
||||
const { server, controller, publisher } = await runSurplusSeller(env);
|
||||
console.log("Surplus seller controller listening");
|
||||
let shuttingDown = false;
|
||||
const shutdown = async (signal: string): Promise<void> => {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
console.log(`Surplus seller controller shutting down (${signal})`);
|
||||
await publisher?.stop();
|
||||
await controller.stop();
|
||||
await new Promise<void>((resolveClose, rejectClose) => {
|
||||
server.close((error) => error ? rejectClose(error) : resolveClose());
|
||||
|
||||
@@ -2,6 +2,7 @@ const USD_RE = /^(?:0|[1-9]\d*)(?:\.\d{1,6})?$/;
|
||||
const EVM_ADDRESS_RE = /^0x[0-9a-fA-F]{40}$/;
|
||||
const POSITIVE_INT_RE = /^[1-9]\d*$/;
|
||||
const NON_NEGATIVE_INT_RE = /^(?:0|[1-9]\d*)$/;
|
||||
const HEX64_RE = /^[0-9a-fA-F]{64}$/;
|
||||
|
||||
export interface SurplusManagerConfig {
|
||||
live: boolean;
|
||||
@@ -25,8 +26,18 @@ export interface SurplusManagerConfig {
|
||||
settledStatuses: string[];
|
||||
statePath: string;
|
||||
port: number;
|
||||
proof: SurplusProofConfig | null;
|
||||
}
|
||||
|
||||
export interface SurplusProofConfig {
|
||||
ogmiosUrl: string;
|
||||
walletPrivateKeyHex: string;
|
||||
ledgerPath: string;
|
||||
reserveLovelace: bigint;
|
||||
feeBudgetLovelace: bigint;
|
||||
intervalMs: number;
|
||||
confirmTimeoutMs: number;
|
||||
}
|
||||
export function loadSurplusManagerConfig(
|
||||
env: Record<string, string | undefined>,
|
||||
): SurplusManagerConfig {
|
||||
@@ -135,6 +146,62 @@ export function loadSurplusManagerConfig(
|
||||
"SURPLUS_STATE_PATH",
|
||||
),
|
||||
port: boundedInteger(env.PORT ?? "8080", "PORT", 1, 65_535),
|
||||
proof: loadProofConfig(env),
|
||||
};
|
||||
}
|
||||
|
||||
function loadProofConfig(
|
||||
env: Record<string, string | undefined>,
|
||||
): SurplusProofConfig | null {
|
||||
const enabledRaw = env.SURPLUS_PROOF_ENABLED;
|
||||
if (enabledRaw === undefined || enabledRaw === "") return null;
|
||||
if (enabledRaw !== "1") {
|
||||
throw new Error("SURPLUS_PROOF_ENABLED must be literal 1 or unset");
|
||||
}
|
||||
|
||||
const ogmiosUrl = required(env, "OGMIOS_URL");
|
||||
const walletPrivateKeyHex = required(
|
||||
env,
|
||||
"SURPLUS_PROOF_WALLET_PRIV_KEY_HEX",
|
||||
);
|
||||
if (!HEX64_RE.test(walletPrivateKeyHex)) {
|
||||
throw new Error(
|
||||
"SURPLUS_PROOF_WALLET_PRIV_KEY_HEX must be 64 hexadecimal characters",
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
ogmiosUrl,
|
||||
walletPrivateKeyHex: walletPrivateKeyHex.toLowerCase(),
|
||||
ledgerPath: requiredWithDefault(
|
||||
env.SURPLUS_PROOF_LEDGER_PATH,
|
||||
"/var/lib/surplus-manager/vector-proofs.json",
|
||||
"SURPLUS_PROOF_LEDGER_PATH",
|
||||
),
|
||||
reserveLovelace: BigInt(boundedInteger(
|
||||
env.SURPLUS_PROOF_RESERVE_LOVELACE ?? "5000000",
|
||||
"SURPLUS_PROOF_RESERVE_LOVELACE",
|
||||
0,
|
||||
Number.MAX_SAFE_INTEGER,
|
||||
)),
|
||||
feeBudgetLovelace: BigInt(boundedInteger(
|
||||
env.SURPLUS_PROOF_FEE_BUDGET_LOVELACE ?? "1000000",
|
||||
"SURPLUS_PROOF_FEE_BUDGET_LOVELACE",
|
||||
1,
|
||||
Number.MAX_SAFE_INTEGER,
|
||||
)),
|
||||
intervalMs: boundedInteger(
|
||||
env.SURPLUS_PROOF_INTERVAL_MS ?? "300000",
|
||||
"SURPLUS_PROOF_INTERVAL_MS",
|
||||
60_000,
|
||||
86_400_000,
|
||||
),
|
||||
confirmTimeoutMs: boundedInteger(
|
||||
env.SURPLUS_PROOF_CONFIRM_TIMEOUT_MS ?? "300000",
|
||||
"SURPLUS_PROOF_CONFIRM_TIMEOUT_MS",
|
||||
10_000,
|
||||
3_600_000,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
268
supplier/src/surplus/vectorPublisher.ts
Normal file
268
supplier/src/surplus/vectorPublisher.ts
Normal file
@@ -0,0 +1,268 @@
|
||||
/**
|
||||
* vectorPublisher.ts — anchors Surplus sale Merkle batches on Vector.
|
||||
*
|
||||
* Policy (2026-08-28): every settled Base USDC settlement transaction gets its
|
||||
* own Vector counterpart. Sales are grouped by their Base `tx_hash`; each group
|
||||
* becomes one Merkle batch anchored in one label-674 Vector transaction.
|
||||
* Observed sales that fell out of the Surplus API window before settling are
|
||||
* anchored together in a trailing catch-up batch so no captured sale is left
|
||||
* without a proof.
|
||||
*
|
||||
* The publisher is fail-closed:
|
||||
* - it never anchors while the wallet cannot cover reserve + fee budget,
|
||||
* - every batch is persisted as a pending intent before any submission,
|
||||
* - a stale pending intent found at cycle start is failed as "submit" and
|
||||
* rebuilt deterministically (same sales ⇒ same root) on the same cycle.
|
||||
*/
|
||||
|
||||
import type { SurplusEarnings, SurplusSale } from "./client.js";
|
||||
import {
|
||||
SURPLUS_SALE_PROOF_PROTOCOL,
|
||||
type CanonicalSurplusSale,
|
||||
type FileSurplusVectorProofLedger,
|
||||
type SurplusPendingBatchIntent,
|
||||
} from "./vectorProof.js";
|
||||
|
||||
export interface SurplusProofAnchorMetadata {
|
||||
p: typeof SURPLUS_SALE_PROOF_PROTOCOL;
|
||||
root: string;
|
||||
count: number;
|
||||
first: string;
|
||||
last: string;
|
||||
}
|
||||
|
||||
export type SurplusProofLedgerSeam = Pick<
|
||||
FileSurplusVectorProofLedger,
|
||||
| "ingestSales"
|
||||
| "selectUnprovedSales"
|
||||
| "persistPendingBatch"
|
||||
| "pendingBatchIntent"
|
||||
| "confirmPendingBatch"
|
||||
| "failPendingBatch"
|
||||
>;
|
||||
|
||||
export interface SurplusVectorProofPublisherOptions {
|
||||
ledger: SurplusProofLedgerSeam;
|
||||
earnings: () => Promise<SurplusEarnings>;
|
||||
anchor: (
|
||||
metadata: SurplusProofAnchorMetadata,
|
||||
) => Promise<{ expectedTxHash: string }>;
|
||||
awaitTx: (txHash: string, timeoutMs: number) => Promise<void>;
|
||||
balanceLovelace: () => Promise<bigint>;
|
||||
reserveLovelace: bigint;
|
||||
feeBudgetLovelace: bigint;
|
||||
settledStatuses: readonly string[];
|
||||
intervalMs: number;
|
||||
confirmTimeoutMs: number;
|
||||
log: (message: string) => void;
|
||||
}
|
||||
|
||||
export interface SurplusPublisherCycleResult {
|
||||
ingested: number;
|
||||
anchored: number;
|
||||
fundingBlocked: boolean;
|
||||
failed: boolean;
|
||||
}
|
||||
|
||||
export class SurplusVectorProofPublisher {
|
||||
private readonly options: SurplusVectorProofPublisherOptions;
|
||||
private timer: NodeJS.Timeout | null = null;
|
||||
private inFlight: Promise<SurplusPublisherCycleResult> | null = null;
|
||||
private stopped = false;
|
||||
|
||||
constructor(options: SurplusVectorProofPublisherOptions) {
|
||||
if (options.reserveLovelace < 0n || options.feeBudgetLovelace <= 0n) {
|
||||
throw new Error(
|
||||
"Surplus proof publisher needs a non-negative reserve and a positive fee budget",
|
||||
);
|
||||
}
|
||||
this.options = options;
|
||||
}
|
||||
|
||||
start(): void {
|
||||
if (this.timer !== null) return;
|
||||
this.stopped = false;
|
||||
void this.tick();
|
||||
this.timer = setInterval(() => void this.tick(), this.options.intervalMs);
|
||||
this.timer.unref?.();
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
if (this.timer !== null) {
|
||||
clearInterval(this.timer);
|
||||
this.timer = null;
|
||||
}
|
||||
if (this.inFlight !== null) {
|
||||
await this.inFlight.catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
private async tick(): Promise<void> {
|
||||
if (this.inFlight !== null || this.stopped) return;
|
||||
const cycle = this.runCycle();
|
||||
this.inFlight = cycle;
|
||||
try {
|
||||
await cycle;
|
||||
} catch (error) {
|
||||
const detail = error instanceof Error ? error.message : String(error);
|
||||
this.options.log(`surplus-proof: cycle failed: ${detail}`);
|
||||
} finally {
|
||||
this.inFlight = null;
|
||||
}
|
||||
}
|
||||
|
||||
async runCycle(): Promise<SurplusPublisherCycleResult> {
|
||||
const { ledger, log } = this.options;
|
||||
const result: SurplusPublisherCycleResult = {
|
||||
ingested: 0,
|
||||
anchored: 0,
|
||||
fundingBlocked: false,
|
||||
failed: false,
|
||||
};
|
||||
|
||||
const settled = await this.observeSettledSales();
|
||||
const { inserted } = await ledger.ingestSales(settled);
|
||||
result.ingested = inserted;
|
||||
if (inserted > 0) {
|
||||
log(`surplus-proof: captured ${inserted} newly settled sale(s)`);
|
||||
}
|
||||
|
||||
const stale = ledger.pendingBatchIntent();
|
||||
if (stale !== null) {
|
||||
await ledger.failPendingBatch(stale.batchRoot, "submit");
|
||||
log(
|
||||
`surplus-proof: failed stale pending batch ${stale.batchRoot} from an interrupted run; retrying this cycle`,
|
||||
);
|
||||
}
|
||||
|
||||
for (const batch of this.planBatches(settled, ledger.selectUnprovedSales())) {
|
||||
if (this.stopped) break;
|
||||
if (!(await this.fundsAvailable())) {
|
||||
result.fundingBlocked = true;
|
||||
break;
|
||||
}
|
||||
const anchored = await this.anchorBatch(batch);
|
||||
if (!anchored) {
|
||||
result.failed = true;
|
||||
break;
|
||||
}
|
||||
result.anchored += 1;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Settled = carries a Base settlement tx hash and passes the status filter. */
|
||||
private async observeSettledSales(): Promise<SurplusSale[]> {
|
||||
const { earnings, settledStatuses } = this.options;
|
||||
const snapshot = await earnings();
|
||||
return snapshot.recentSales.filter((sale) =>
|
||||
sale.transactionHash !== null &&
|
||||
(settledStatuses.length === 0 ||
|
||||
settledStatuses.includes(sale.settlementStatus))
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* One batch per observed Base settlement tx (oldest first), then one
|
||||
* catch-up batch for unproved sales whose settlement tx is no longer
|
||||
* observable in the Surplus API window.
|
||||
*/
|
||||
private planBatches(
|
||||
settled: readonly SurplusSale[],
|
||||
unproved: readonly CanonicalSurplusSale[],
|
||||
): CanonicalSurplusSale[][] {
|
||||
const txBySaleId = new Map<string, string>();
|
||||
for (const sale of settled) {
|
||||
if (sale.transactionHash !== null) {
|
||||
txBySaleId.set(sale.id, sale.transactionHash);
|
||||
}
|
||||
}
|
||||
|
||||
const groups = new Map<string, CanonicalSurplusSale[]>();
|
||||
const leftovers: CanonicalSurplusSale[] = [];
|
||||
for (const sale of unproved) {
|
||||
const settlementTx = txBySaleId.get(sale.saleId);
|
||||
if (settlementTx === undefined) {
|
||||
leftovers.push(sale);
|
||||
continue;
|
||||
}
|
||||
const group = groups.get(settlementTx);
|
||||
if (group === undefined) groups.set(settlementTx, [sale]);
|
||||
else group.push(sale);
|
||||
}
|
||||
|
||||
const batches = [...groups.values()].sort(
|
||||
(left, right) => earliest(left) - earliest(right),
|
||||
);
|
||||
if (leftovers.length > 0) batches.push(leftovers);
|
||||
return batches;
|
||||
}
|
||||
|
||||
private async fundsAvailable(): Promise<boolean> {
|
||||
const { balanceLovelace, reserveLovelace, feeBudgetLovelace, log } =
|
||||
this.options;
|
||||
const balance = await balanceLovelace();
|
||||
if (balance >= reserveLovelace + feeBudgetLovelace) return true;
|
||||
log(
|
||||
`surplus-proof: funding required — wallet holds ${balance} lovelace, ` +
|
||||
`needs reserve ${reserveLovelace} + fee budget ${feeBudgetLovelace}; anchoring paused`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
private async anchorBatch(
|
||||
sales: readonly CanonicalSurplusSale[],
|
||||
): Promise<boolean> {
|
||||
const { ledger, anchor, awaitTx, confirmTimeoutMs, log } = this.options;
|
||||
const saleIds = sales.map((sale) => sale.saleId).sort();
|
||||
const intent: SurplusPendingBatchIntent = await ledger.persistPendingBatch(
|
||||
saleIds,
|
||||
);
|
||||
|
||||
let expectedTxHash: string;
|
||||
try {
|
||||
const built = await anchor({
|
||||
p: SURPLUS_SALE_PROOF_PROTOCOL,
|
||||
root: intent.batchRoot,
|
||||
count: intent.count,
|
||||
first: intent.firstSaleId,
|
||||
last: intent.lastSaleId,
|
||||
});
|
||||
expectedTxHash = built.expectedTxHash;
|
||||
} catch (error) {
|
||||
await ledger.failPendingBatch(intent.batchRoot, "submit");
|
||||
const detail = error instanceof Error ? error.message : String(error);
|
||||
log(
|
||||
`surplus-proof: submit failed for batch ${intent.batchRoot} (${intent.count} sale(s)): ${detail}`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
await awaitTx(expectedTxHash, confirmTimeoutMs);
|
||||
} catch (error) {
|
||||
await ledger.failPendingBatch(intent.batchRoot, "confirm");
|
||||
const detail = error instanceof Error ? error.message : String(error);
|
||||
log(
|
||||
`surplus-proof: confirmation timed out for batch ${intent.batchRoot} in tx ${expectedTxHash}: ${detail}`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
await ledger.confirmPendingBatch(intent.batchRoot, expectedTxHash);
|
||||
log(
|
||||
`surplus-proof: anchored ${intent.count} sale(s) [${intent.firstSaleId}..${intent.lastSaleId}] ` +
|
||||
`root ${intent.batchRoot} in Vector tx ${expectedTxHash}`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
function earliest(sales: readonly CanonicalSurplusSale[]): number {
|
||||
let value = Number.POSITIVE_INFINITY;
|
||||
for (const sale of sales) {
|
||||
if (sale.createdAtEpochMs < value) value = sale.createdAtEpochMs;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
Reference in New Issue
Block a user