Tickets Pool

package.json

@@ -56,6 +56,7 @@
     "packages/jam/state-json",
     "packages/jam/state-merkleization",
     "packages/jam/state-vectors",
+    "packages/jam/ticket-pool",
     "packages/jam/transition",
     "packages/jam/transition/disputes",
     "packages/misc/benchmark",

packages/jam/jamnp-s/package.json

@@ -21,6 +21,7 @@
     "@typeberry/logger": "*",
     "@typeberry/networking": "*",
     "@typeberry/numbers": "*",
+    "@typeberry/ticket-pool": "*",
     "@typeberry/utils": "*"
   }
 }

packages/jam/jamnp-s/tasks/ticket-distribution.test.ts

@@ -8,7 +8,8 @@ import { tinyChainSpec } from "@typeberry/config";
 import { BANDERSNATCH_PROOF_BYTES } from "@typeberry/crypto";
 import { Logger } from "@typeberry/logger";
 import { createTestPeerPair, MockNetwork } from "@typeberry/networking/testing.js";
-import { OK } from "@typeberry/utils";
+import { AcceptTicketsValidator, ValidationError } from "@typeberry/ticket-pool";
+import { OK, Result } from "@typeberry/utils";
 import { Connections } from "../peers.js";
 import { StreamManager } from "../stream-manager.js";
 import { TicketDistributionTask } from "./ticket-distribution.js";
@@ -41,6 +42,10 @@ describe("TicketDistributionTask", () => {
     // Use real TicketDistributionTask
     const ticketTask = TicketDistributionTask.start(streamManager, connections, tinyChainSpec);
 
+    // Default validator accepts every ticket so the test asserts purely on distribution
+    // behaviour. Tests that exercise the rejection path overwrite this.
+    ticketTask.setTicketValidator(new AcceptTicketsValidator());
+
     // Intercept received tickets by wrapping onTicketReceived behavior
     // The task already adds received tickets to pending queue via addTicket,
     // so we can track them by checking the pending queue growth or by
@@ -273,7 +278,7 @@ describe("TicketDistributionTask", () => {
     assert.deepStrictEqual(peer2.receivedTickets[0].ticket, ticket);
   });
 
-  it("should NOT redistribute ticket if validation callback returns false", async () => {
+  it("should NOT redistribute ticket if validator rejects", async () => {
     const self = await init("self");
     const peer1 = await init("peer1");
     const peer2 = await init("peer2");
@@ -283,21 +288,23 @@ describe("TicketDistributionTask", () => {
     await tick();
 
     // Validation always rejects
-    self.ticketTask.setOnTicketReceived(async () => false);
+    self.ticketTask.setTicketValidator({
+      validate: async () => Result.error(ValidationError.InvalidProof, () => "rejected"),
+    });
 
     const ticket = createTestTicket(0);
     peer1.ticketTask.addTicket(TEST_EPOCH, ticket);
     peer1.ticketTask.maintainDistribution();
     await tick();
 
-    // self.addTicket was NOT called (callback returned false), so nothing to redistribute
+    // self.addTicket was NOT called (validator rejected), so nothing to redistribute
     assert.strictEqual(self.receivedTickets.length, 0);
     self.ticketTask.maintainDistribution();
     await tick();
     assert.strictEqual(peer2.receivedTickets.length, 0);
   });
 
-  it("should redistribute ticket if validation callback returns true", async () => {
+  it("should redistribute ticket if validator accepts", async () => {
     const self = await init("self");
     const peer1 = await init("peer1");
     const peer2 = await init("peer2");
@@ -306,19 +313,40 @@ describe("TicketDistributionTask", () => {
     self.openConnection(peer2);
     await tick();
 
-    // Validation always accepts
-    self.ticketTask.setOnTicketReceived(async () => true);
-
+    // Default init() already wires an AcceptTicketsValidator
     const ticket = createTestTicket(0);
     peer1.ticketTask.addTicket(TEST_EPOCH, ticket);
     peer1.ticketTask.maintainDistribution();
     await tick();
 
-    // self.addTicket WAS called (callback returned true)
+    // self.addTicket WAS called
     assert.strictEqual(self.receivedTickets.length, 1);
     self.ticketTask.maintainDistribution();
     await tick();
     assert.strictEqual(peer2.receivedTickets.length, 1);
     assert.deepStrictEqual(peer2.receivedTickets[0].ticket, ticket);
   });
+
+  it("replacePool overwrites the redistribution pool", async () => {
+    const self = await init("self");
+    const peer1 = await init("peer1");
+
+    self.openConnection(peer1);
+    await tick();
+
+    // Locally added tickets first
+    self.ticketTask.addTicket(TEST_EPOCH, createTestTicket(0));
+    self.ticketTask.addTicket(TEST_EPOCH, createTestTicket(1));
+
+    // Pool dump replaces with a different set
+    const dump = [createTestTicket(2), createTestTicket(3)];
+    self.ticketTask.replacePool(TEST_EPOCH, dump);
+
+    self.ticketTask.maintainDistribution();
+    await tick();
+
+    assert.strictEqual(peer1.receivedTickets.length, 2);
+    assert.deepStrictEqual(peer1.receivedTickets[0].ticket, dump[0]);
+    assert.deepStrictEqual(peer1.receivedTickets[1].ticket, dump[1]);
+  });
 });

packages/jam/jamnp-s/tasks/ticket-distribution.ts

@@ -2,6 +2,7 @@ import type { Epoch } from "@typeberry/block";
 import type { SignedTicket } from "@typeberry/block/tickets.js";
 import type { ChainSpec } from "@typeberry/config";
 import { Logger } from "@typeberry/logger";
+import { DenyTicketsValidator, PendingTicketPool, type TicketValidator } from "@typeberry/ticket-pool";
 import { OK } from "@typeberry/utils";
 import type { AuxData, Connections } from "../peers.js";
 import { ce131 } from "../protocol/index.js";
@@ -26,6 +27,11 @@ const TICKET_AUX: AuxData<TicketAuxData> = {
  * Uses CE-132 (proxy-to-all) for direct broadcast to all peers.
  * Implements a maintain pattern similar to SyncTask: tickets are collected
  * and periodically distributed to peers that haven't received them yet.
+ *
+ * Incoming tickets from peers are first run through a {@link TicketValidator};
+ * only validated tickets are added to the redistribution pool. The default
+ * validator denies everything, so callers must wire a real one via
+ * {@link setTicketValidator} before any networked ticket can be redistributed.
  */
 export class TicketDistributionTask {
   static start(streamManager: StreamManager, connections: Connections, chainSpec: ChainSpec) {
@@ -44,10 +50,8 @@ export class TicketDistributionTask {
     return task;
   }
 
-  /** Pending tickets waiting to be distributed to peers */
-  private pendingTickets: Array<{ epochIndex: Epoch; ticket: SignedTicket }> = [];
-  /** Current epoch being tracked (cleared when epoch changes) */
-  private currentEpoch: Epoch | null = null;
+  private readonly pool = new PendingTicketPool();
+  private validator: TicketValidator = new DenyTicketsValidator();
 
   private constructor(
     private readonly streamManager: StreamManager,
@@ -58,16 +62,14 @@ export class TicketDistributionTask {
    * Should be called periodically to distribute pending tickets to connected peers.
    */
   maintainDistribution() {
-    if (this.currentEpoch === null) {
-      return; // No tickets to distribute yet
+    const currentEpoch = this.pool.currentEpoch;
+    if (currentEpoch === null) {
+      return;
     }
 
-    /** `this` is mutable and TS can't narrow this.currentEpoch inside the callback closure */
-    const currentEpoch = this.currentEpoch;
-
-    // Iterate through all pending tickets
-    for (let ticketIdx = 0; ticketIdx < this.pendingTickets.length; ticketIdx++) {
-      const { epochIndex, ticket } = this.pendingTickets[ticketIdx];
+    const tickets = this.pool.getTickets();
+    for (let ticketIdx = 0; ticketIdx < tickets.length; ticketIdx++) {
+      const { epochIndex, ticket } = tickets[ticketIdx];
 
       // Try to send to each connected peer
       for (const peerInfo of this.connections.getConnectedPeers()) {
@@ -108,79 +110,47 @@ export class TicketDistributionTask {
   }
 
   /**
-   * Add a ticket to the pending queue for distribution.
+   * Add a ticket to the redistribution pool.
    * Clears pending tickets when epoch changes.
    * Deduplicates tickets based on signature.
    */
   addTicket(epochIndex: Epoch, ticket: SignedTicket) {
-    // Drop tickets for older epochs (can happen when a delayed validation callback completes
-    // after the epoch has already advanced — accepting it would roll back currentEpoch).
-    if (this.currentEpoch !== null && epochIndex < this.currentEpoch) {
-      return;
-    }
-
-    // Epoch advanced — clear old tickets
-    if (this.currentEpoch !== null && epochIndex > this.currentEpoch) {
-      logger.log`[addTicket] Epoch changed from ${this.currentEpoch} to ${epochIndex}, clearing ${this.pendingTickets.length} old tickets`;
-      this.pendingTickets = [];
-      // Note: We don't need to clear aux data for all peers here.
-      // The aux data contains the epoch, so maintainDistribution will lazily
-      // reset it when it detects an epoch mismatch. This handles both connected
-      // and disconnected peers correctly.
-    }
-
-    this.currentEpoch = epochIndex;
-
-    /**
-     * Deduplicate: check if a ticket with the same signature already exists
-     *
-     * Here we are risking "poisoning" the local pendingTickets - i.e:
-     *  1. The adversary sees a signature and swaps the ticket attempt to something different.
-     *  2. This creates an invalid ticket, but prevents a valid ticket with the same signature from being included and distributed.
-     *
-     * TODO [MaSi]: The poisoning risk should be fixed during implementation of ticket validation.
-     */
-    const isDuplicate = this.pendingTickets.some(
-      (pending) => pending.epochIndex === epochIndex && pending.ticket.signature.isEqualTo(ticket.signature),
-    );
-
-    if (!isDuplicate) {
-      this.pendingTickets.push({ epochIndex, ticket });
-      logger.info`[addTicket] Added ticket for epoch ${epochIndex}, total: ${this.pendingTickets.length}`;
-    }
+    this.pool.addTicket(epochIndex, ticket);
   }
 
-  private onTicketReceivedCallback: ((epochIndex: Epoch, ticket: SignedTicket) => Promise<boolean>) | null = null;
+  /**
+   * Replace the redistribution pool for the given epoch with the supplied tickets.
+   * Used when the authorship worker dumps the authoritative pool on an epoch boundary.
+   */
+  replacePool(epochIndex: Epoch, tickets: readonly SignedTicket[]) {
+    this.pool.replace(epochIndex, tickets);
+  }
 
   /**
-   * Register a callback that validates a received ticket.
-   * The ticket is only added to the redistribution pool if the callback returns `true`.
-   * This prevents redistribution of invalid tickets (e.g. those with a tampered `attempt` field).
+   * Register the validator that decides whether tickets received from peers should be
+   * accepted (and therefore redistributed). The default is {@link DenyTicketsValidator},
+   * so the caller must install a real validator for any peer ticket to make it through.
    */
-  setOnTicketReceived(cb: (epochIndex: Epoch, ticket: SignedTicket) => Promise<boolean>) {
-    this.onTicketReceivedCallback = cb;
+  setTicketValidator(validator: TicketValidator) {
+    this.validator = validator;
   }
 
   private onTicketReceived(epochIndex: Epoch, ticket: SignedTicket) {
     logger.trace`Received ticket for epoch ${epochIndex}, attempt ${ticket.attempt}`;
-    if (this.onTicketReceivedCallback !== null) {
-      // Validate first; only redistribute if valid to avoid spreading tampered tickets.
-      // Wrap with Promise.resolve().then() to catch both sync throws and async rejections.
-      const cb = this.onTicketReceivedCallback;
-      Promise.resolve()
-        .then(() => cb(epochIndex, ticket))
-        .then((isValid) => {
-          if (isValid) {
-            this.addTicket(epochIndex, ticket);
-          } else {
-            logger.warn`Dropping invalid ticket for epoch ${epochIndex} (validation failed)`;
-          }
-        })
-        .catch((error) => {
-          logger.error`Error validating ticket for epoch ${epochIndex}, attempt ${ticket.attempt}: ${error}`;
-        });
-    } else {
-      this.addTicket(epochIndex, ticket);
-    }
+    const validator = this.validator;
+    // Wrap with Promise.resolve().then() so a synchronous throw inside the validator
+    // funnels into the same .catch() as an async rejection.
+    Promise.resolve()
+      .then(() => validator.validate(epochIndex, ticket))
+      .then((result) => {
+        if (result.isOk) {
+          this.addTicket(epochIndex, ticket);
+        } else {
+          logger.trace`Dropping ticket for epoch ${epochIndex}: ${result.error}`;
+        }
+      })
+      .catch((error) => {
+        logger.error`Error validating ticket for epoch ${epochIndex}, attempt ${ticket.attempt}: ${error}`;
+      });
   }
 }

packages/jam/ticket-pool/index.ts

@@ -0,0 +1,3 @@
+export * from "./pending-ticket-pool.js";
+export * from "./ticket-validator.js";
+export * from "./verified-ticket-pool.js";

packages/jam/ticket-pool/package.json

@@ -0,0 +1,21 @@
+{
+  "name": "@typeberry/ticket-pool",
+  "version": "0.6.0",
+  "description": "In-memory Safrole ticket pools and validation abstraction shared between block authorship and networking.",
+  "main": "index.ts",
+  "dependencies": {
+    "@typeberry/block": "*",
+    "@typeberry/bytes": "*",
+    "@typeberry/collections": "*",
+    "@typeberry/crypto": "*",
+    "@typeberry/hash": "*",
+    "@typeberry/logger": "*",
+    "@typeberry/utils": "*"
+  },
+  "scripts": {
+    "test": "tsx --test $(find . -type f -name '*.test.ts' | tr '\\n' ' ')"
+  },
+  "author": "Fluffy Labs",
+  "license": "MPL-2.0",
+  "type": "module"
+}