diff --git a/src/node.ts b/src/node.ts index afba90d..aad4545 100644 --- a/src/node.ts +++ b/src/node.ts @@ -3,18 +3,22 @@ import { encodeMessage, decodeLine } from "./protocol/codec.js"; import type { NodeConfig } from "./config.js"; import { computeObjectId } from "./protocol/hashing.js"; import { isMissingObjectStoreError, ObjectStore } from "./store/objectStore.js"; +import { Mempool as TransactionMempool } from "./store/mempool.js"; import { PeerStore } from "./store/peerStore.js"; import { type ApplicationObject, type AnyMessage, + type BlockWithMetadata, type ErrorMessage, type GetObjectMessage, type IHaveObjectMessage, + type GetMemPool, + type Mempool as MempoolMessage, type ObjectMessage, + type Transaction, type UtxoSnapshot, - GENESIS_BLOCK, - GetChainTipMessage, - ChainTip + type GetChainTipMessage, + type ChainTip } from "./types.js"; import { ApplicationObjectValidationError, @@ -37,15 +41,63 @@ interface ConnectionState { interface ObjectWaiter { resolve: (object: ApplicationObject) => void; reject: (error: Error) => void; + timeout: NodeJS.Timeout | null; +} + +// Compares two chain tips and returns txids removed from the old fork and added on the new fork. +async function collectForkDiff( + oldTipId: string, + newTipId: string, + store: ObjectStore +): Promise<{ orphaned: string[]; confirmed: string[] }> { + // Loads one block object and rejects corrupted or mistyped store entries. + const getBlock = async (blockId: string): Promise => { + const object = await store.getObject(blockId); + if (object.type !== "blockwithmetadata") { + throw new Error(`Object ${blockId} is not a block`); + } + return object; + }; + + let oldCursor = await getBlock(oldTipId); + let newCursor = await getBlock(newTipId); + const oldBlocks: BlockWithMetadata[] = []; + const newBlocks: BlockWithMetadata[] = []; + + // First move the taller chain back until both cursors are at the same height. + while (oldCursor.height > newCursor.height) { + oldBlocks.push(oldCursor); + oldCursor = await getBlock(oldCursor.block.previd!); + } + + while (newCursor.height > oldCursor.height) { + newBlocks.push(newCursor); + newCursor = await getBlock(newCursor.block.previd!); + } + + // Then walk both chains together; collected old blocks are orphaned, new blocks are confirmed. + while (computeObjectId(oldCursor) !== computeObjectId(newCursor)) { + oldBlocks.push(oldCursor); + newBlocks.push(newCursor); + oldCursor = await getBlock(oldCursor.block.previd!); + newCursor = await getBlock(newCursor.block.previd!); + } + + return { + orphaned: oldBlocks.flatMap((block) => block.block.txids), + confirmed: newBlocks.flatMap((block) => block.block.txids) + }; } export class MarabuNode { private readonly server: net.Server; private readonly connections = new Map(); private readonly pendingObjectWaiters = new Map>(); + private readonly inFlightObjectValidations = new Map>(); private readonly outboundSocketsByPeer = new Map(); private readonly dialingPeers = new Set(); private readonly failedAttemptsByPeer = new Map(); + private readonly mempool = new TransactionMempool(); private readonly maxFailedAttempts = 3; private reconnectTimer: NodeJS.Timeout | null = null; @@ -116,6 +168,12 @@ export class MarabuNode { return await this.handleGetChaintipMessage(socket, message); case "chaintip": return await this.handleChaintipMessage(socket, message); + case "getmempool": + this.handleGetMempoolMessage(socket, message); + return true; + case "mempool": + await this.handleMempoolMessage(message); + return true; default: { // Reject unexpected validated variants defensively. this.sendErrorAndClose(socket, { @@ -155,26 +213,36 @@ export class MarabuNode { socket: net.Socket, message: ObjectMessage ): Promise { + const incomingObjectId = computeObjectId(message.object); + try { - const objectToSave = await validateApplicationObjectState(message.object, { - getObject: (key: string) => this.objectStore.getObject(key), - getUtxo: (blockId: string) => this.objectStore.getUtxo(blockId), - putUtxo: (blockId: string, snapshot: UtxoSnapshot) => this.objectStore.putUtxo(blockId, snapshot), - requestObject: (objectId: string) => this.sendGetObjectToAllPeers(objectId), - waitForObject: (objectId: string, timeoutMs: number) => this.waitForObject(objectId, timeoutMs) - }); - const objectId = computeObjectId(objectToSave); - if (await this.objectStore.hasObject(objectId)) { + if (await this.objectStore.hasObject(incomingObjectId)) { return true; } - await this.objectStore.putObject(objectId, objectToSave); - if (objectToSave.type === "blockwithmetadata") { - await this.updateChainTip({ type: "chaintip", blockid: objectId }); + const inFlightValidation = this.inFlightObjectValidations.get(incomingObjectId); + if (inFlightValidation !== undefined) { + await inFlightValidation; + return true; } - this.resolveObjectWaiters(objectId, objectToSave); - this.sendIHaveObjectToAllPeers(objectId); + this.markObjectArrived(incomingObjectId); + + const validation = this.validateAndStoreObject(incomingObjectId, message.object) + .catch((error: unknown) => { + const failure = error instanceof Error ? error : new Error(String(error)); + this.rejectObjectWaiters( + incomingObjectId, + this.createObjectDependencyError(incomingObjectId, failure) + ); + throw failure; + }) + .finally(() => { + this.inFlightObjectValidations.delete(incomingObjectId); + }); + + this.inFlightObjectValidations.set(incomingObjectId, validation); + await validation; return true; } catch (error) { if (error instanceof MissingParentBlockError) { @@ -200,6 +268,51 @@ export class MarabuNode { } } + // Validates incoming objects, stores accepted ones, and applies chain/mempool side effects. + private async validateAndStoreObject( + objectId: string, + object: ApplicationObject + ): Promise { + const objectToSave = await validateApplicationObjectState(object, { + getObject: (key: string) => this.objectStore.getObject(key), + getUtxo: (blockId: string) => this.objectStore.getUtxo(blockId), + putUtxo: (blockId: string, snapshot: UtxoSnapshot) => this.objectStore.putUtxo(blockId, snapshot), + requestObject: (requestedObjectId: string) => this.sendGetObjectToAllPeers(requestedObjectId), + waitForObject: (requestedObjectId: string, timeoutMs: number) => this.waitForObject(requestedObjectId, timeoutMs) + }); + + if (await this.objectStore.hasObject(objectId)) { + const storedObject = await this.objectStore.getObject(objectId); + this.resolveObjectWaiters(objectId, storedObject); + return storedObject; + } + + await this.objectStore.putObject(objectId, objectToSave); + if (objectToSave.type === "transaction" && !("height" in objectToSave)) { + // Store valid transaction objects even when they do not fit the active-chain mempool. + if (!this.mempool.tryAddTransaction(objectId, objectToSave)) { + this.resolveObjectWaiters(objectId, objectToSave); + throw new ApplicationObjectValidationError( + "INVALID_TX_OUTPOINT", + `Transaction ${objectId} cannot be applied to the active mempool` + ); + } + } + + if (objectToSave.type === "blockwithmetadata") { + // Keep the previous tip so a real tip change can remove confirmed txs and restore reorg orphans. + const previousTipId = await this.objectStore.getChainTip(); + const tipChanged = await this.updateChainTip({ type: "chaintip", blockid: objectId }); + if (tipChanged) { + await this.rebuildMempoolAfterBlock(objectToSave, previousTipId); + } + } + + this.resolveObjectWaiters(objectId, objectToSave); + this.sendIHaveObjectToAllPeers(objectId); + return objectToSave; + } + // Handles an ihaveobject message from a peer. private async handleIHaveObjectMessage( socket: net.Socket, @@ -238,7 +351,9 @@ export class MarabuNode { } } - private async updateChainTip(chaintip: ChainTip): Promise { + // Updates the stored chain tip if the candidate block is strictly better. + private async updateChainTip(chaintip: ChainTip): Promise { + // Fetch the candidate block - must be a valid blockwithmetadata object. const newTipObject = await this.objectStore.getObject(chaintip.blockid); if (newTipObject.type !== "blockwithmetadata") { throw new Error(`Chain tip candidate ${chaintip.blockid} is not a block`); @@ -246,23 +361,98 @@ export class MarabuNode { let currentTipBlockId: string; try { + // Try to retrieve the current chain tip's block id from the object store. currentTipBlockId = await this.objectStore.getChainTip(); } catch (error: unknown) { + // If there is no chain tip stored yet accept the candidate as the new tip. if (isMissingObjectStoreError(error)) { await this.objectStore.putChainTip(chaintip.blockid); - return; + return true; } - + // Other errors should propagate. throw error; } + // Fetch the current tip block object for comparison. const currentTipObject = await this.objectStore.getObject(currentTipBlockId); if (currentTipObject.type !== "blockwithmetadata") { + // Corrupted store: chain tip does not refer to a block. throw new Error(`Stored chain tip ${currentTipBlockId} is not a block`); } + // If the candidate's height is higher, update the tip. if (newTipObject.height > currentTipObject.height) { await this.objectStore.putChainTip(chaintip.blockid); + return true; + } + + // Otherwise, keep the current tip. + return false; + } + + + // Rebuilds the in-memory mempool after a new active-chain block is accepted. + private async rebuildMempoolAfterBlock( + newBlock: BlockWithMetadata, + previousTipId: string + ): Promise { + const newBlockId = computeObjectId(newBlock); + + const confirmedTxids = new Set(newBlock.block.txids); + const orphanedTxids: string[] = []; + + if (newBlock.block.previd !== previousTipId) { + // Reorg case: old-chain txs may become mempool candidates again. + const forkDiff = await collectForkDiff(previousTipId, newBlockId, this.objectStore); + for (const txid of forkDiff.confirmed) { + confirmedTxids.add(txid); + } + orphanedTxids.push(...forkDiff.orphaned); + } + + const survivingTxids = this.mempool + .getTxids() + .filter((txid) => !confirmedTxids.has(txid)); + const candidateTxids = [ + ...survivingTxids, + ...orphanedTxids.filter((txid) => !confirmedTxids.has(txid)) + ]; + const candidates: { txid: string; transaction: Transaction }[] = []; + + // Reload candidates from storage so rebuild only replays transaction objects we still have. + for (const txid of candidateTxids) { + try { + const object = await this.objectStore.getObject(txid); + if (object.type === "transaction" && !("height" in object)) { + candidates.push({ txid, transaction: object }); + } + } catch (error: unknown) { + if (!isMissingObjectStoreError(error)) { + throw error; + } + } + } + + const newUtxo = await this.objectStore.getUtxo(newBlockId); + // Reset to confirmed chain state, then replay surviving/orphaned transactions that still fit. + this.mempool.rebuild(newUtxo, candidates); + } + + + // Handles a getmempool message from a peer. + private handleGetMempoolMessage(socket: net.Socket, _message: GetMemPool): void { + this.sendMessage(socket, { + type: "mempool", + txids: this.mempool.getTxids() + }); + } + + // Requests mempool transactions that this node has not seen yet. + private async handleMempoolMessage(message: MempoolMessage): Promise { + for (const txid of message.txids) { + if (!await this.objectStore.hasObject(txid)) { + this.sendGetObjectToAllPeers(txid); + } } } @@ -311,6 +501,15 @@ export class MarabuNode { } } + // Sends a getmempool request to all connected peers. + private sendGetMempoolToAllPeers(): void { + for (const socket of this.connections.keys()) { + this.sendMessage(socket, { + type: "getmempool" + }); + } + } + // Sends a getobject request to a connected peer. private sendGetObjectToPeer(socket: net.Socket, objectId: string): void { this.sendMessage(socket, { @@ -350,13 +549,20 @@ export class MarabuNode { }, timeoutMs); const waiter: ObjectWaiter = { + timeout, resolve: (object) => { - clearTimeout(timeout); + if (waiter.timeout !== null) { + clearTimeout(waiter.timeout); + waiter.timeout = null; + } removeWaiter(); resolve(object); }, reject: (error) => { - clearTimeout(timeout); + if (waiter.timeout !== null) { + clearTimeout(waiter.timeout); + waiter.timeout = null; + } removeWaiter(); reject(error); } @@ -365,6 +571,10 @@ export class MarabuNode { waiters.add(waiter); this.pendingObjectWaiters.set(objectId, waiters); + if (this.inFlightObjectValidations.has(objectId)) { + this.markObjectArrived(objectId); + } + // Re-check after registering the waiter so we do not miss a just-arrived object. void this.objectStore .getObject(objectId) @@ -382,6 +592,21 @@ export class MarabuNode { }); } + // Clears arrival timers after the exact requested object is being validated. + private markObjectArrived(objectId: string): void { + const waiters = this.pendingObjectWaiters.get(objectId); + if (waiters === undefined) { + return; + } + + for (const waiter of waiters) { + if (waiter.timeout !== null) { + clearTimeout(waiter.timeout); + waiter.timeout = null; + } + } + } + // Resolves all waiters that were blocked on the given object ID. private resolveObjectWaiters(objectId: string, object: ApplicationObject): void { const waiters = this.pendingObjectWaiters.get(objectId); @@ -408,6 +633,14 @@ export class MarabuNode { } } + // Wraps a validation failure so dependent block waits can map it to UNFINDABLE_OBJECT. + private createObjectDependencyError(objectId: string, cause: Error): Error { + const error = new Error(`Object ${objectId} failed validation while resolving a dependency`); + error.name = "ObjectDependencyValidationError"; + error.cause = cause; + return error; + } + // Rejects every outstanding object wait when the node is shutting down. private rejectAllPendingObjectWaiters(error: Error): void { for (const objectId of this.pendingObjectWaiters.keys()) { @@ -451,6 +684,11 @@ export class MarabuNode { this.server.listen(this.config.port, this.config.host); }); + const tipId = await this.objectStore.getChainTip(); + const tipUtxo = await this.objectStore.getUtxo(tipId); + // The mempool is intentionally in-memory, so rebuild its base view on every boot. + this.mempool.initialize(tipUtxo); + this.running = true; // Attempt immediate outbound dials instead of waiting for the first interval tick. this.tryConnectDiscoveredPeers(); @@ -461,9 +699,10 @@ export class MarabuNode { }, this.config.reconnectIntervalMs); - // Wait briefly to allow outbound peer connections, then send getchaintip to all. + // Wait briefly to allow outbound peer connections, then ask for chain and mempool state. setTimeout(() => { this.sendGetChaintipToAllPeers(); + this.sendGetMempoolToAllPeers(); }, 500); // 500ms delay; adjust as needed based on network conditions } diff --git a/src/store/mempool.ts b/src/store/mempool.ts new file mode 100644 index 0000000..8b7b08f --- /dev/null +++ b/src/store/mempool.ts @@ -0,0 +1,87 @@ +import type { Transaction, UtxoEntry, UtxoSnapshot } from "../types.js"; + +function utxoKey(txid: string, index: number): string { + return `${txid}:${index}`; +} + +export class Mempool { + private readonly txids = new Set(); + private readonly utxoMap = new Map(); + + initialize(snapshot: UtxoSnapshot): void { + this.txids.clear(); + this.utxoMap.clear(); + + for (const entry of snapshot.entries) { + this.utxoMap.set(utxoKey(entry.outpoint.txid, entry.outpoint.index), entry); + } + } + + getUtxoMap(): Map { + return this.utxoMap; + } + + getTxids(): string[] { + return [...this.txids]; + } + + has(txid: string): boolean { + return this.txids.has(txid); + } + + tryAddTransaction(txid: string, transaction: Transaction): boolean { + if (!this.canApply(transaction)) { + return false; + } + + this.addTransaction(txid, transaction); + return true; + } + + addTransaction(txid: string, transaction: Transaction): void { + this.txids.add(txid); + + for (const input of transaction.inputs) { + this.utxoMap.delete(utxoKey(input.outpoint.txid, input.outpoint.index)); + } + + for (const [index, output] of transaction.outputs.entries()) { + this.utxoMap.set(utxoKey(txid, index), { + outpoint: { txid, index }, + output + }); + } + } + + rebuild( + snapshot: UtxoSnapshot, + candidates: { txid: string; transaction: Transaction }[] + ): void { + this.initialize(snapshot); + + const remaining = [...candidates]; + let appliedAny = true; + + while (appliedAny) { + appliedAny = false; + + for (let index = 0; index < remaining.length; index += 1) { + const candidate = remaining[index]; + if (candidate === undefined || !this.canApply(candidate.transaction)) { + continue; + } + + this.addTransaction(candidate.txid, candidate.transaction); + remaining.splice(index, 1); + index -= 1; + appliedAny = true; + } + } + } + + private canApply(transaction: Transaction): boolean { + return transaction.inputs.every((input) => + this.utxoMap.has(utxoKey(input.outpoint.txid, input.outpoint.index)) + ); + } +} diff --git a/src/types.ts b/src/types.ts index afebd85..bff021c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -41,14 +41,14 @@ export interface IHaveObjectMessage { objectid: string; } -// export interface GetMemPool { -// type: "getmempool"; -// } +export interface GetMemPool { + type: "getmempool"; +} -// export interface Mempool { -// type: "mempool"; -// txids: string[]; -// } +export interface Mempool { + type: "mempool"; + txids: string[]; +} export interface GetChainTipMessage { type: "getchaintip"; @@ -149,6 +149,6 @@ export type AnyMessage = | IHaveObjectMessage | ObjectMessage | GetChainTipMessage - | ChainTip; - // | GetMemPool - // | Mempool + | ChainTip + | GetMemPool + | Mempool; diff --git a/src/validation/messageSchema.ts b/src/validation/messageSchema.ts index 1215d5b..7cb4efc 100644 --- a/src/validation/messageSchema.ts +++ b/src/validation/messageSchema.ts @@ -15,7 +15,9 @@ import type { OutPoint, Block, GetChainTipMessage, - ChainTip + ChainTip, + GetMemPool, + Mempool } from "../types.js"; import { ValidationError, @@ -59,6 +61,10 @@ export function validateWireMessage(value: unknown): AnyMessage { return validateGetChainTipMessage(record); case "chaintip": return validateChainTipMessage(record); + case "getmempool": + return validateGetMempoolMessage(record); + case "mempool": + return validateMempoolMessage(record); default: throw new MessageValidationError("Unknown message type"); } @@ -232,13 +238,33 @@ function validateTransactionMessage(value: RecordValue): Transaction { ); } + // Parse and validate each input object. This ensures every input in the array is a JSON object + // and properly shaped per the Input schema. + const inputs = value.inputs.map((input, index) => + validateInput( + assertRecord(input, `transaction.inputs[${index}] must be a JSON object`) + ) + ); + + // Check for duplicate outpoints within the transaction inputs, + // which is a protocol-level stateless violation. + const seen = new Set(); + for (const input of inputs) { + // Form a unique key for each outpoint using its txid and index. + const key = `${input.outpoint.txid}:${input.outpoint.index}`; + if (seen.has(key)) { + // If the same outpoint is referenced more than once in the inputs, reject the message. + throw new MessageValidationError( + `transaction.inputs contains duplicate outpoint ${key}` + ); + } + seen.add(key); + } + + // Construct and return the normalized Transaction return { type: "transaction", - inputs: value.inputs.map((input, index) => - validateInput( - assertRecord(input, `transaction.inputs[${index}] must be a JSON object`) - ) - ), + inputs, outputs: value.outputs.map((output, index) => validateOutput( assertRecord(output, `transaction.outputs[${index}] must be a JSON object`) @@ -363,6 +389,31 @@ function validateChainTipMessage(value: RecordValue): ChainTip { }; } +// Validates a "getmempool" request message. +function validateGetMempoolMessage(value: RecordValue): GetMemPool { + assertExactKeys(value, ["type"]); + + return { + type: "getmempool" + }; +} + +// Validates a "mempool" response message. +function validateMempoolMessage(value: RecordValue): Mempool { + assertExactKeys(value, ["type", "txids"]); + + if (!Array.isArray(value.txids)) { + throw new MessageValidationError("mempool.txids must be an array"); + } + + return { + type: "mempool", + txids: value.txids.map((txid, index) => + assertObjectId(txid, `mempool.txids[${index}]`) + ) + }; +} + /*////////////////////////////////////////////////////////////// SPECIFIC HELPERS //////////////////////////////////////////////////////////////*/ diff --git a/src/validation/objectState.ts b/src/validation/objectState.ts index 100c075..63b84d8 100644 --- a/src/validation/objectState.ts +++ b/src/validation/objectState.ts @@ -18,7 +18,6 @@ import { isValidEd25519PublicKey } from "./utils.js"; import { computeObjectId } from "../protocol/hashing.js"; -import { error } from "node:console"; const REQUIRED_BLOCK_TARGET = "00000000abc00000000000000000000000000000000000000000000000000000"; @@ -295,7 +294,6 @@ async function validateTransactionState( validateTransactionConservation(transaction, referencedOutputs); } - // Resolves each input's outpoint to the referenced output, performing existence and type checks. async function resolveOutpoints( transaction: Transaction, @@ -346,6 +344,7 @@ async function resolveOutpoints( return referencedOutputs; } + // Validates that each output has a valid Ed25519 pubkey and a non-negative integer value. function validateOutputs(outputs: unknown): void { const context = "transaction"; @@ -479,6 +478,11 @@ function isObjectWaitTimeoutError(error: unknown): boolean { return error instanceof Error && error.name === "ObjectWaitTimeoutError"; } +// Returns true when a matching object arrived but failed validation before storage. +function isObjectDependencyValidationError(error: unknown): boolean { + return error instanceof Error && error.name === "ObjectDependencyValidationError"; +} + // Loads a block transaction immediately or waits for it to arrive after requesting it. async function findBlockTransactionObject( txid: string, @@ -499,7 +503,8 @@ async function findBlockTransactionObject( } catch (error: unknown) { if ( !isMissingReferencedObjectError(error) && - !isObjectWaitTimeoutError(error) + !isObjectWaitTimeoutError(error) && + !isObjectDependencyValidationError(error) ) { throw error; } @@ -607,7 +612,8 @@ async function checkPrevId(block: Block, objectLookup: ObjectLookup): Promise { port, lines: [helloLine, JSON.stringify({ type: "object", object: vectors.gossipTx })] }); - assertNoError(gossipResponses, "gossip transaction"); - - const ihave = await peer.waitFor( - (message) => - typeof message === "object" && - message !== null && - "type" in message && - (message as Record).type === "ihaveobject" && - (message as Record).objectid === vectors.gossipTxId + assertError( + gossipResponses, + "INVALID_TX_OUTPOINT", + "transaction outside active mempool" + ); + + await assert.rejects( + peer.waitFor( + (message) => + typeof message === "object" && + message !== null && + "type" in message && + (message as Record).type === "ihaveobject" && + (message as Record).objectid === vectors.gossipTxId, + 500 + ), + /timeout/ ); - assert.ok(ihave, "expected ihaveobject gossip to peer"); } finally { peer.socket.end(); } diff --git a/test/pset2/tx-validation.test.ts b/test/pset2/tx-validation.test.ts index 82e1622..cb81f1f 100644 --- a/test/pset2/tx-validation.test.ts +++ b/test/pset2/tx-validation.test.ts @@ -65,6 +65,7 @@ async function buildVectors(): Promise<{ coinbase: Record; coinbaseId: string; validTx: Record; + validTxId: string; invalidSigTx: Record; invalidOutpointTx: Record; unknownObjectTx: Record; @@ -93,6 +94,7 @@ async function buildVectors(): Promise<{ }; const validTx = await makeTx(coinbaseId, 0, 10); + const validTxId = objectId(validTx); const invalidOutpointTx = await makeTx(coinbaseId, 1, 10); const unknownObjectTx = await makeTx("00".repeat(32), 0, 10); const invalidConservationTx = await makeTx(coinbaseId, 0, 60); @@ -110,6 +112,7 @@ async function buildVectors(): Promise<{ coinbase, coinbaseId, validTx, + validTxId, invalidSigTx, invalidOutpointTx, unknownObjectTx, @@ -134,7 +137,22 @@ test("PSET2: transaction validation", async () => { port, lines: [helloLine, JSON.stringify({ type: "object", object: vectors.validTx })] }); - assertNoError(validResponses, "valid transaction"); + assertError( + validResponses, + "INVALID_TX_OUTPOINT", + "valid transaction outside active mempool" + ); + + const getStoredTxResponses = await sendLines({ + host, + port, + lines: [ + helloLine, + JSON.stringify({ type: "getobject", objectid: vectors.validTxId }) + ] + }); + const storedTx = findMessage(getStoredTxResponses, "object"); + assert.ok(storedTx, "expected mempool-rejected valid transaction to be stored"); const unknownResponses = await sendLines({ host, diff --git a/test/pset3/block-validation.test.ts b/test/pset3/block-validation.test.ts index b3b9982..71ea0e8 100644 --- a/test/pset3/block-validation.test.ts +++ b/test/pset3/block-validation.test.ts @@ -1,5 +1,6 @@ import assert from "node:assert/strict"; import { test } from "node:test"; +import { setTimeout as delay } from "node:timers/promises"; import { computeObjectId } from "../../src/protocol/hashing.js"; import type { ApplicationObject, @@ -52,6 +53,13 @@ const ASSIGNMENT_BLOCK: Block = { const ASSIGNMENT_BLOCK_ID = computeObjectId(ASSIGNMENT_BLOCK); +interface NodeWaiterInternals { + waitForObject: (objectId: string, timeoutMs: number) => Promise; + markObjectArrived: (objectId: string) => void; + resolveObjectWaiters: (objectId: string, object: ApplicationObject) => void; + inFlightObjectValidations: Map>; +} + // Checks that a response set contains no protocol error. function assertNoError( messages: HarnessParsedMessage[], @@ -312,6 +320,75 @@ test("PSET3: valid block is accepted and gossiped with ihaveobject", async () => }); }); +test("PSET3: object waiter timeout pauses after matching object arrives for validation", async () => { + await withTestNode(async ({ node }) => { + const internals = node as unknown as NodeWaiterInternals; + const objectId = computeObjectId(ASSIGNMENT_TX); + let settled = false; + + const wait = internals.waitForObject(objectId, 50).finally(() => { + settled = true; + }); + + await delay(10); + internals.markObjectArrived(objectId); + await delay(100); + assert.equal(settled, false, "waiter should not time out after object arrival"); + + internals.resolveObjectWaiters(objectId, ASSIGNMENT_TX); + assert.deepEqual(await wait, ASSIGNMENT_TX); + }); +}); + +test("PSET3: waiter registered during in-flight validation does not start its timeout", async () => { + await withTestNode(async ({ node }) => { + const internals = node as unknown as NodeWaiterInternals; + const objectId = computeObjectId(ASSIGNMENT_TX); + let settled = false; + let resolveValidation!: (object: ApplicationObject) => void; + const validation = new Promise((resolve) => { + resolveValidation = resolve; + }); + + internals.inFlightObjectValidations.set(objectId, validation); + try { + const wait = internals.waitForObject(objectId, 50).finally(() => { + settled = true; + }); + + await delay(100); + assert.equal(settled, false, "waiter should remain pending while validation is in flight"); + + resolveValidation(ASSIGNMENT_TX); + internals.resolveObjectWaiters(objectId, ASSIGNMENT_TX); + assert.deepEqual(await wait, ASSIGNMENT_TX); + } finally { + internals.inFlightObjectValidations.delete(objectId); + } + }); +}); + +test("PSET3: dependency validation failure maps to UNFINDABLE_OBJECT", async () => { + const dependencyError = new Error("dependency failed validation"); + dependencyError.name = "ObjectDependencyValidationError"; + + const objectLookup = createObjectLookup({ + objects: new Map() + }); + + await assertValidationError( + ASSIGNMENT_BLOCK, + "UNFINDABLE_OBJECT", + { + ...objectLookup, + waitForObject: async () => { + throw dependencyError; + } + }, + "dependency validation failure" + ); +}); + test("PSET3: block transaction that spends outside the parent UTXO set returns INVALID_TX_OUTPOINT", async () => { const regularTx: Transaction = { type: "transaction",