PoC: Race condition prevention in worker

AGENTS.md

@@ -6,6 +6,7 @@
 It uses [Bun](https://bun.com/) for JavaScript and Typescript runtime and package management (instead of Nodejs + npm).
 
 The content is organized into a Bun workspace. See @README.md for:
+
 - High-level architecture and component interactions
 - Detailed setup instructions
 - Functional flow documentation
@@ -15,13 +16,13 @@
 
 All packages are in the `./packages` directory:
 
-| Package | Type | Purpose |
-|---------|------|---------|
-| `btcindexer` | Service (Worker) | Bitcoin-to-Sui bridging and minting |
-| `sui-indexer` | Service (Worker) | Sui blockchain monitoring and redemption |
-| `block-ingestor` | Service (Worker) | Receives Bitcoin blocks via REST API |
-| `compliance` | Service (Worker) | Sanctions and geo-blocking data |
-| `lib` | Shared Library | Common utilities and types |
+| Package          | Type             | Purpose                                  |
+| ---------------- | ---------------- | ---------------------------------------- |
+| `btcindexer`     | Service (Worker) | Bitcoin-to-Sui bridging and minting      |
+| `sui-indexer`    | Service (Worker) | Sui blockchain monitoring and redemption |
+| `block-ingestor` | Service (Worker) | Receives Bitcoin blocks via REST API     |
+| `compliance`     | Service (Worker) | Sanctions and geo-blocking data          |
+| `lib`            | Shared Library   | Common utilities and types               |
 
 ### Core Technologies
 
@@ -138,7 +139,7 @@
 
 **Location**: `./packages/sui-indexer`
 
 ### Architecture
 
 - `src/index.ts` - Entry point with scheduled task
 - `src/processor.ts` - Sui event indexing
@@ -156,13 +157,13 @@
 - Manages presign objects for Bitcoin transaction signing
 - Implements coin selection logic for redemption transactions
 
 ### Key Features
 
 1. **Event Monitoring**: Indexes Sui events for nBTC redemption requests
 2. **Redemption Processing**: Handles burn-and-redeem flow with IKA MPC
 3. **UTXO Management**: Manages UTXO lifecycle (available → locked → spent)
 
 ### Configuration
 
 - **Cron**: Every minute (`* * * * *`)
 - **D1 Database**: Shared `btcindexer-dev`
@@ -176,18 +177,18 @@
 
 See @packages/block-ingestor/README.md for detailed architecture.
 
 ### Architecture
 
 - `src/index.ts` - HTTP router and handlers
 - `src/ingest.ts` - Block ingestion logic
 - `src/api/put-blocks.ts` - msgpack encoding/decoding
 - `src/api/client.ts` - Client for sending blocks
 
 ### Key Features
 
 Receives Bitcoin blocks via REST API, validates them, and enqueues to `block-queue` for processing by BTCIndexer.
 
 ### Configuration
 
 - **KV Namespace**: `BtcBlocks` (shared with btcindexer)
 - **Queue Producer**: `block-queue`
@@ -197,25 +198,25 @@
 
 **Location**: `./packages/compliance`
 
 ### Architecture
 
 - `src/index.ts` - Scheduled worker entry point
 - `src/sanction.ts` - Sanctions list updating logic
 - `src/storage.ts` - D1 storage for sanctions
 - `src/rpc.ts` - RPC interface for other services to query compliance data
 
 ### Key Features
 
 1. **Sanctions List Updates**: Daily cron job fetches and updates sanctions data
 2. **Compliance API**: Exposes RPC methods for other services to check addresses
 3. **Geo-blocking**: Supports geo-blocking rules
 
 ### Configuration
 
 - **Cron**: Daily at 1am (`0 1 * * *`)
 - **D1 Database**: `compliance`
 
 ### Database Schema
 
 See migration files in `packages/compliance/db/migrations/`.
 

packages/btcindexer/db/migrations/0001_initial_schema.sql

@@ -126,3 +126,9 @@ CREATE TABLE IF NOT EXISTS presign_objects (
 ) STRICT;
 
 CREATE INDEX IF NOT EXISTS presign_objects_sui_network_created_at ON presign_objects(sui_network, created_at);
+
+CREATE TABLE IF NOT EXISTS cron_locks (
+	lock_name TEXT NOT NULL PRIMARY KEY,
+	acquired_at INTEGER NOT NULL,
+	expires_at INTEGER NOT NULL
+) STRICT;

packages/lib/src/test-helpers/init_db.ts

@@ -42,6 +42,7 @@ export const tables = [
 	"btc_blocks",
 	"indexer_state",
 	"presign_objects",
+	"cron_locks",
 	"setups",
 ];
 

packages/sui-indexer/src/index.ts

@@ -25,19 +25,35 @@ export default {
 	},
 	async scheduled(_event: ScheduledController, env: Env, _ctx: ExecutionContext): Promise<void> {
 		const storage = new D1Storage(env.DB);
-		const activeNetworks = await storage.getActiveNetworks();
-
-		const mnemonic = await getSecret(env.NBTC_MINTING_SIGNER_MNEMONIC);
-		const suiClients = await createSuiClients(activeNetworks, mnemonic);
-
-		// Run both indexer and redeem solver tasks in parallel
-		const results = await Promise.allSettled([
-			runSuiIndexer(storage, activeNetworks, suiClients),
-			runRedeemSolver(storage, env, suiClients, activeNetworks),
-		]);
+		const lockToken = await storage.acquireLock("sui-indexer-cron", 5 * 60 * 1000); // 5 minutes
+		if (lockToken === null) {
+			logger.warn({
+				msg: "Cron job already running, skipping this execution",
+				lockName: "sui-indexer-cron",
+			});
+			return;
+		}
 
-		// Check for any rejected promises and log errors
-		reportErrors(results, "scheduled", "Scheduled task error", ["SuiIndexer", "RedeemSolver"]);
+		try {
+			const activeNetworks = await storage.getActiveNetworks();
+
+			const mnemonic = await getSecret(env.NBTC_MINTING_SIGNER_MNEMONIC);
+			const suiClients = await createSuiClients(activeNetworks, mnemonic);
+
+			// Run both indexer and redeem solver tasks in parallel
+			const results = await Promise.allSettled([
+				runSuiIndexer(storage, activeNetworks, suiClients),
+				runRedeemSolver(storage, env, suiClients, activeNetworks),
+			]);
+
+			// Check for any rejected promises and log errors
+			reportErrors(results, "scheduled", "Scheduled task error", [
+				"SuiIndexer",
+				"RedeemSolver",
+			]);
+		} finally {
+			await storage.releaseLock("sui-indexer-cron", lockToken);
+		}
 	},
 } satisfies ExportedHandler<Env>;
 

packages/sui-indexer/src/storage.test.ts

@@ -597,4 +597,85 @@ describe("IndexerStorage", () => {
 		expect(inputs[0]!.input_index).toBe(0);
 		expect(inputs[1]!.input_index).toBe(1);
 	});
+
+	describe("Distributed Lock", () => {
+		it("should acquire lock when none exists", async () => {
+			const token = await storage.acquireLock("test-lock", 60000);
+			expect(token).not.toBeNull();
+
+			const lock = await db
+				.prepare("SELECT * FROM cron_locks WHERE lock_name = ?")
+				.bind("test-lock")
+				.first<{ lock_name: string }>();
+			expect(lock).not.toBeNull();
+			expect(lock!.lock_name).toBe("test-lock");
+		});
+
+		it("should fail to acquire lock when already held (not expired)", async () => {
+			const first = await storage.acquireLock("test-lock", 60000);
+			expect(first).not.toBeNull();
+
+			const second = await storage.acquireLock("test-lock", 60000);
+			expect(second).toBeNull();
+		});
+
+		it("should acquire lock when existing lock is expired", async () => {
+			const expiredTime = Date.now() - 10000;
+			await db
+				.prepare(
+					"INSERT INTO cron_locks (lock_name, acquired_at, expires_at) VALUES (?, ?, ?)",
+				)
+				.bind("test-lock", expiredTime - 60000, expiredTime)
+				.run();
+
+			const token = await storage.acquireLock("test-lock", 60000);
+			expect(token).not.toBeNull();
+
+			const lock = await db
+				.prepare("SELECT * FROM cron_locks WHERE lock_name = ?")
+				.bind("test-lock")
+				.first<{ expires_at: number }>();
+			expect(lock!.expires_at).toBeGreaterThan(Date.now());
+		});
+
+		it("should release lock with matching token", async () => {
+			const token = await storage.acquireLock("test-lock", 60000);
+			expect(token).not.toBeNull();
+
+			await storage.releaseLock("test-lock", token!);
+
+			const lock = await db
+				.prepare("SELECT * FROM cron_locks WHERE lock_name = ?")
+				.bind("test-lock")
+				.first();
+			expect(lock).toBeNull();
+		});
+
+		it("should allow reacquiring lock after release", async () => {
+			const first = await storage.acquireLock("test-lock", 60000);
+			expect(first).not.toBeNull();
+
+			await storage.releaseLock("test-lock", first!);
+
+			const second = await storage.acquireLock("test-lock", 60000);
+			expect(second).not.toBeNull();
+		});
+
+		it("should not release another instance's lock after expiry", async () => {
+			const tokenA = await storage.acquireLock("test-lock", 10);
+			expect(tokenA).not.toBeNull();
+			await new Promise((resolve) => setTimeout(resolve, 20));
+			const tokenB = await storage.acquireLock("test-lock", 60000);
+			expect(tokenB).not.toBeNull();
+
+			await storage.releaseLock("test-lock", tokenA!);
+
+			const lock = await db
+				.prepare("SELECT * FROM cron_locks WHERE lock_name = ?")
+				.bind("test-lock")
+				.first<{ acquired_at: number }>();
+			expect(lock).not.toBeNull();
+			expect(lock!.acquired_at).toBe(tokenB!);
+		});
+	});
 });

packages/sui-indexer/src/storage.ts

@@ -854,6 +854,34 @@ export class D1Storage {
 			sui_network: toSuiNet(result.sui_network),
 		};
 	}
+
+	async acquireLock(lockName: string, ttlMs: number): Promise<number | null> {
+		const now = Date.now();
+		try {
+			const result = await this.db
+				.prepare(
+					`INSERT INTO cron_locks (lock_name, acquired_at, expires_at)
+					 VALUES (?, ?, ?)
+					 ON CONFLICT(lock_name) DO UPDATE
+					 SET acquired_at = excluded.acquired_at, expires_at = excluded.expires_at
+					 WHERE cron_locks.expires_at <= excluded.acquired_at
+					 RETURNING acquired_at`,
+				)
+				.bind(lockName, now, now + ttlMs)
+				.first<number>("acquired_at");
+			return result ?? null;
+		} catch (error) {
+			logError({ msg: "Failed to acquire lock", method: "acquireLock", lockName }, error);
+			return null;
+		}
+	}
+
+	async releaseLock(lockName: string, acquiredAt: number): Promise<void> {
+		await this.db
+			.prepare(`DELETE FROM cron_locks WHERE lock_name = ? AND acquired_at = ?`)
+			.bind(lockName, acquiredAt)
+			.run();
+	}
 }
 
 export async function insertRedeemRequest(