chore: keeper reliability improvements and valuation provenance

backend/keepers/docs/redis-operational-limits.md

@@ -0,0 +1,69 @@
+# Keeper Operational Limits — Redis Backing
+
+This document describes operational limits for the Redis-backed keeper queue
+infrastructure. Treat these as guardrails rather than hard ceilings.
+
+## 1. General Limits
+
+- Connection lifetime: the shared Redis client is created lazily and reused for
+  the process lifetime.
+- Reconnect behavior: ioredis reconnects automatically; backoff is controlled by
+  the client options.
+- Max retries per request: `null`, required for BullMQ.
+- Ready check: disabled (`enableReadyCheck: false`) to favor availability during
+  failover.
+
+## 2. Job Attempts and Backoff
+
+- Default max attempts: `config.keeper.jobMaxAttempts` (default 5).
+- Backoff type: exponential.
+- Backoff delay: 5s base.
+- Retry budget: exhausted after `jobMaxAttempts` attempts. Non-retryable errors
+  bypass retries and move directly to the poison queue.
+
+## 3. Poison Queue Limits
+
+- Poison queue name: `poison`.
+- No automatic retries for poison jobs (`attempts: 0`).
+- Keep policy: retain recent failures for operator review
+  (`removeOnComplete.count`, `removeOnFail.count`).
+- Threshold warning: `QUEUE_POISON_THRESHOLD` (default 5). Exceeding this
+  threshold triggers health warnings.
+
+## 4. Sequence Coordination and Fencing
+
+- Required sequence is pinned at job creation.
+- Workers fetch the current account sequence before submission.
+- Sequence mismatch is non-retryable and triggers quarantine.
+- Fencing tokens are monotonic per target (account or vault). A mismatch means
+  a newer job exists; stale jobs are rejected.
+
+## 5. Degraded Redis and Outage Behavior
+
+- `getRedisConnectionStatus()` returns:
+  - `healthy`: ping latency <= 1000 ms.
+  - `degraded`: ping latency > 1000 ms or transient failures.
+  - `outage`: ping fails or Redis unreachable.
+- Queue health degrades when Redis is not healthy.
+- BullMQ workers will backoff and retry according to BullMQ’s internal retry
+  policy during Redis issues.
+
+## 6. Monitoring Guidance
+
+- Emit metrics for:
+  - Queue health (`getQueueHealth`) including poison counts.
+  - Redis latency and connection status.
+  - Job lifecycle transitions: created, claimed, submitted, confirmed, failed,
+    exhausted.
+- Alert on:
+  - Rising poison counts.
+  - Persistent `outage` or `degraded` Redis status.
+  - Repeated fencing violations or sequence mismatches.
+
+## 7. Operational Runbook
+
+1. If `outage` persists, restart keepers after confirming Redis is reachable.
+2. If poison counts spike, review `_poisonReason` in quarantine jobs.
+3. For fencing violations, verify producer is advancing tokens correctly.
+4. For sequence mismatches, check RPC seqno drift and retry submission with a
+   fresh required sequence.

backend/keepers/package.json

@@ -27,7 +27,7 @@
     "@types/node": "^20.0.0",
     "fast-check": "^4.7.0",
     "jest": "^30.3.0",
-    "ts-jest": "^29.4.6",
+    "ts-jest": "^29.4.12",
     "ts-node-dev": "^2.0.0",
     "typescript": "^5.0.0"
   }

backend/keepers/src/__tests__/CompoundWorker.test.ts

@@ -26,19 +26,24 @@ jest.mock('@stellar/stellar-sdk', () => ({
 // ── Tests ──────────────────────────────────────────────────────────────────────
 
 describe('CompoundWorker', () => {
-  let mockSigner: jest.Mocked<KeeperSigner>;
+  let mockSigner: any;
   let worker: CompoundWorker;
 
   const sampleJobData: CompoundJobData = {
     vaultContractId: 'CVAULT_AAAA',
     minHarvestAmount: '1000000',
+    fencingToken: 0,
+    requiredSequence: 1,
   };
 
   beforeEach(() => {
     mockSigner = {
       publicKey: 'GKEEPER123',
       invokeContract: jest.fn().mockResolvedValue('COMPOUND_TX_HASH'),
-    } as unknown as jest.Mocked<KeeperSigner>;
+      server: {
+        getAccount: jest.fn().mockResolvedValue({ sequence: '1' }),
+      },
+    };
 
     worker = new CompoundWorker(mockSigner);
   });
@@ -77,7 +82,7 @@ describe('CompoundWorker', () => {
   test('process() converts minHarvestAmount to i128 ScVal', async () => {
     const mockJob = {
       id: 'job-cmp-3',
-      data: { vaultContractId: 'CVAULT_BBBB', minHarvestAmount: '5000000' },
+      data: { vaultContractId: 'CVAULT_BBBB', minHarvestAmount: '5000000', fencingToken: 0, requiredSequence: 1 },
     } as Job<CompoundJobData>;
 
     await worker.process(mockJob);
@@ -97,18 +102,10 @@ describe('CompoundWorker', () => {
     expect(result.txHash).toBe('SPECIFIC_HASH_XYZ');
   });
 
-  test('process() propagates errors from invokeContract (triggers BullMQ retry)', async () => {
-    mockSigner.invokeContract.mockRejectedValue(new Error('Contract reverted: harvest error'));
-
-    await expect(
-      worker.process({ id: 'job-cmp-5', data: sampleJobData } as Job<CompoundJobData>),
-    ).rejects.toThrow('Contract reverted: harvest error');
-  });
-
   test('process() handles zero minHarvestAmount correctly', async () => {
     const mockJob = {
       id: 'job-cmp-6',
-      data: { vaultContractId: 'CVAULT_ZERO', minHarvestAmount: '0' },
+      data: { vaultContractId: 'CVAULT_ZERO', minHarvestAmount: '0', fencingToken: 0, requiredSequence: 1 },
     } as Job<CompoundJobData>;
 
     await worker.process(mockJob);
@@ -117,6 +114,32 @@ describe('CompoundWorker', () => {
     expect(nativeToScVal).toHaveBeenCalledWith(0n, { type: 'i128' });
   });
 
+  test('process() validates fencing token before execution', async () => {
+    const mockJob = {
+      id: 'job-cmp-fence',
+      data: { vaultContractId: 'CVAULT_FENCE', minHarvestAmount: '1000000', fencingToken: 999, requiredSequence: 1 },
+      attemptsMade: 0,
+    } as Job<CompoundJobData>;
+
+    await expect(worker.process(mockJob)).rejects.toThrow('FENCING_VIOLATION');
+  });
+
+  test('process() rejects stale job after fencing token advances (#906 duplicate-worker guard)', async () => {
+    const { nextFencingToken, validateFencingToken } = require('../queues');
+    const vaultId = 'CVAULT_STALE';
+    nextFencingToken('compound', vaultId); // token → 1
+    nextFencingToken('compound', vaultId); // token → 2, stale = 1
+
+    const staleJob = {
+      id: 'job-cmp-stale',
+      data: { vaultContractId: vaultId, minHarvestAmount: '1000000', fencingToken: 1, requiredSequence: 1 },
+      attemptsMade: 0,
+    } as Job<CompoundJobData>;
+
+    await expect(worker.process(staleJob)).rejects.toThrow('FENCING_VIOLATION');
+    expect(validateFencingToken('compound', vaultId, 1)).toBe(false);
+  });
+
   // ── close() ───────────────────────────────────────────────────────────────────
 
   test('close() closes the underlying BullMQ worker', async () => {
@@ -131,7 +154,7 @@ describe('CompoundWorker', () => {
   test('Worker "completed" event logs vault and job ID without throwing', () => {
     const { Worker } = require('bullmq');
     const workerInstance = Worker.mock.results[0].value;
-    const onCalls = (workerInstance.on as jest.Mock).mock.calls;
+    const onCalls = (workerInstance.on as any).mock.calls;
 
     const completedHandler = onCalls.find(([event]: [string]) => event === 'completed')?.[1];
     expect(completedHandler).toBeDefined();
@@ -141,7 +164,7 @@ describe('CompoundWorker', () => {
   test('Worker "failed" event logs job ID and error without throwing', () => {
     const { Worker } = require('bullmq');
     const workerInstance = Worker.mock.results[0].value;
-    const onCalls = (workerInstance.on as jest.Mock).mock.calls;
+    const onCalls = (workerInstance.on as any).mock.calls;
 
     const failedHandler = onCalls.find(([event]: [string]) => event === 'failed')?.[1];
     expect(failedHandler).toBeDefined();