diff --git a/backend/keepers/docs/redis-operational-limits.md b/backend/keepers/docs/redis-operational-limits.md new file mode 100644 index 000000000..28a804c86 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/backend/keepers/package-lock.json b/backend/keepers/package-lock.json index 1f8acd8ea..d36d8cd2b 100644 --- a/backend/keepers/package-lock.json +++ b/backend/keepers/package-lock.json @@ -23,7 +23,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" } @@ -1633,6 +1633,9 @@ "arm64" ], "dev": true, + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -1647,6 +1650,43 @@ "arm64" ], "dev": true, + "libc": [ + "musl" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@unrs/resolver-binding-linux-loong64-gnu": { + "version": "1.12.2", + "resolved": "https://registry.npmjs.org/@unrs/resolver-binding-linux-loong64-gnu/-/resolver-binding-linux-loong64-gnu-1.12.2.tgz", + "integrity": "sha512-3SJGEh1DborhG6pyxvhPzCT4bbSIVihsvgJc13P1bHG7KLdNDaF9T3gsTwFc7Jw/5Y5/iWOjkEx7Zy0NvCGX3Q==", + "cpu": [ + "loong64" + ], + "dev": true, + "libc": [ + "glibc" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@unrs/resolver-binding-linux-loong64-musl": { + "version": "1.12.2", + "resolved": "https://registry.npmjs.org/@unrs/resolver-binding-linux-loong64-musl/-/resolver-binding-linux-loong64-musl-1.12.2.tgz", + "integrity": "sha512-jiuG/Obbel7uw1PwHNFfrkiKhLAF6mnyZ6aWlOAVN9WqKm8v0OFGnciJIHu8+CMvXLQ8AD51LPzAoUfT21D5Ew==", + "cpu": [ + "loong64" + ], + "dev": true, + "libc": [ + "musl" + ], "license": "MIT", "optional": true, "os": [ @@ -1689,6 +1729,9 @@ "ppc64" ], "dev": true, + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -1703,6 +1746,9 @@ "riscv64" ], "dev": true, + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -1717,6 +1763,9 @@ "riscv64" ], "dev": true, + "libc": [ + "musl" + ], "license": "MIT", "optional": true, "os": [ @@ -1731,6 +1780,9 @@ "s390x" ], "dev": true, + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -1745,6 +1797,9 @@ "x64" ], "dev": true, + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -1759,6 +1814,9 @@ "x64" ], "dev": true, + "libc": [ + "musl" + ], "license": "MIT", "optional": true, "os": [ diff --git a/backend/keepers/package.json b/backend/keepers/package.json index d81017ecc..9f9ce6770 100644 --- a/backend/keepers/package.json +++ b/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" } diff --git a/backend/keepers/src/__tests__/CompoundWorker.test.ts b/backend/keepers/src/__tests__/CompoundWorker.test.ts index 45915daba..0a6d80782 100644 --- a/backend/keepers/src/__tests__/CompoundWorker.test.ts +++ b/backend/keepers/src/__tests__/CompoundWorker.test.ts @@ -26,19 +26,24 @@ jest.mock('@stellar/stellar-sdk', () => ({ // ── Tests ────────────────────────────────────────────────────────────────────── describe('CompoundWorker', () => { - let mockSigner: jest.Mocked; + 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; + 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; 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), - ).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; 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; + + 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; + + 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(); diff --git a/backend/keepers/src/__tests__/LiquidationWorker.test.ts b/backend/keepers/src/__tests__/LiquidationWorker.test.ts index 72767c8fc..a73778cb5 100644 --- a/backend/keepers/src/__tests__/LiquidationWorker.test.ts +++ b/backend/keepers/src/__tests__/LiquidationWorker.test.ts @@ -10,36 +10,47 @@ jest.mock('../utils/redis', () => ({ })); jest.mock('bullmq', () => ({ - Worker: jest.fn().mockImplementation((_name, _processor, _opts) => ({ + Worker: jest.fn().mockImplementation((_name: string, _processor: unknown, _opts: unknown) => ({ on: jest.fn(), close: jest.fn().mockResolvedValue(undefined), })), + Queue: jest.fn().mockImplementation((name: string) => ({ + name, + add: jest.fn().mockResolvedValue({ id: 'poison-job' }), + close: jest.fn().mockResolvedValue(undefined), + })), })); jest.mock('@stellar/stellar-sdk', () => ({ - Address: jest.fn().mockImplementation((addr) => ({ + Address: jest.fn().mockImplementation((addr: string) => ({ toScVal: jest.fn().mockReturnValue({ type: 'address', value: addr }), })), + nativeToScVal: jest.fn().mockReturnValue({ type: 'i128', value: 0n }), })); // ── Tests ────────────────────────────────────────────────────────────────────── describe('LiquidationWorker', () => { - let mockSigner: jest.Mocked; + let mockSigner: any; let worker: LiquidationWorker; const sampleJobData: LiquidationJobData = { - accountAddress: 'GUNDERCOLLATERALIZED', - currentCrBps: 9500, - collateralValueUsd: '100000', - debtAmount: '50000', + accountAddress: 'GACCOUNT_123', + currentCrBps: 10500, + collateralValueUsd: '1000000000', + debtAmount: '500000000', + fencingToken: 0, + requiredSequence: 42, }; beforeEach(() => { mockSigner = { publicKey: 'GKEEPER123', - invokeContract: jest.fn().mockResolvedValue('TX_HASH_ABC123'), - } as unknown as jest.Mocked; + invokeContract: jest.fn().mockResolvedValue('LIQUIDATION_TX_HASH'), + server: { + getAccount: jest.fn().mockResolvedValue({ sequence: '42' }), + }, + }; worker = new LiquidationWorker(mockSigner); }); @@ -48,38 +59,52 @@ describe('LiquidationWorker', () => { jest.clearAllMocks(); }); - test('process() calls invokeContract with correct method and args', async () => { + // ── process() ──────────────────────────────────────────────────────────────── + + test('process() calls invokeContract with "liquidate" method and correct contract', async () => { const mockJob = { - id: '1', + id: 'job-liq-1', data: sampleJobData, + attemptsMade: 0, } as Job; const result = await worker.process(mockJob); expect(mockSigner.invokeContract).toHaveBeenCalledWith( - expect.any(String), // contract ID from config + expect.any(String), 'liquidate', expect.arrayContaining([expect.anything(), expect.anything()]), expect.objectContaining({ workerName: 'LiquidationWorker', jobId: '1' }), ); - expect(result).toEqual({ txHash: 'TX_HASH_ABC123' }); + expect(result).toEqual({ txHash: 'LIQUIDATION_TX_HASH' }); }); - test('process() returns the transaction hash on success', async () => { - mockSigner.invokeContract.mockResolvedValue('DEADBEEF_TX_HASH'); + test('process() passes keeper public key as first arg', async () => { + const mockJob = { id: 'job-liq-2', data: sampleJobData, attemptsMade: 0 } as Job; + await worker.process(mockJob); - const mockJob = { id: '2', data: sampleJobData } as Job; - const result = await worker.process(mockJob); - - expect(result.txHash).toBe('DEADBEEF_TX_HASH'); + const { Address } = require('@stellar/stellar-sdk'); + expect(Address).toHaveBeenCalledWith('GKEEPER123'); }); - test('process() propagates errors from invokeContract (triggers BullMQ retry)', async () => { - mockSigner.invokeContract.mockRejectedValue(new Error('Simulation failed')); + test('process() verifies Stellar sequence before submission', async () => { + const mockJob = { + id: 'job-liq-4', + data: { ...sampleJobData, requiredSequence: 999 }, + attemptsMade: 0, + } as Job; - const mockJob = { id: '3', data: sampleJobData } as Job; + mockSigner.server.getAccount = jest.fn().mockResolvedValue({ sequence: '42' }); - await expect(worker.process(mockJob)).rejects.toThrow('Simulation failed'); + await expect(worker.process(mockJob)).rejects.toThrow('SEQUENCE_MISMATCH'); + }); + + test('process() propagates errors from invokeContract (triggers retry/quarantine)', async () => { + mockSigner.invokeContract.mockRejectedValue(new Error('Contract reverted: liquidation error')); + + await expect( + worker.process({ id: 'job-liq-5', data: sampleJobData, attemptsMade: 0 } as Job), + ).rejects.toThrow('Contract reverted: liquidation error'); }); // ── Dry-run policy fixtures (issue #986) ──────────────────────────────── @@ -177,28 +202,14 @@ describe('LiquidationWorker', () => { expect(workerInstance.close).toHaveBeenCalled(); }); - // ── Event callbacks ────────────────────────────────────────────────────────── - - test('Worker "completed" event logs the job ID and account address', () => { - const { Worker } = require('bullmq'); - const workerInstance = Worker.mock.results[0].value; - const onCalls = (workerInstance.on as jest.Mock).mock.calls; - - const completedHandler = onCalls.find(([event]: [string]) => event === 'completed')?.[1]; - expect(completedHandler).toBeDefined(); - // Should not throw when invoked with a completed job - expect(() => completedHandler({ id: 'j1', data: sampleJobData })).not.toThrow(); - }); - - test('Worker "failed" event logs the job ID and error', () => { + 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(); - // Should not throw even when called with null job (e.g. stalled jobs) - expect(() => failedHandler(null, new Error('timeout'))).not.toThrow(); - expect(() => failedHandler({ id: 'j2', data: sampleJobData }, new Error('rpc error'))).not.toThrow(); + expect(() => failedHandler(null, new Error('liquidation failed'))).not.toThrow(); + expect(() => failedHandler({ id: 'lj2', data: sampleJobData }, new Error('undercollateralized'))).not.toThrow(); }); -}); +}); \ No newline at end of file diff --git a/backend/keepers/src/__tests__/queues.test.ts b/backend/keepers/src/__tests__/queues.test.ts index 6a37461b3..8af9b6a18 100644 --- a/backend/keepers/src/__tests__/queues.test.ts +++ b/backend/keepers/src/__tests__/queues.test.ts @@ -94,6 +94,147 @@ describe('queues/index', () => { }); }); +describe('queues — exactly-once fencing (#906)', () => { + beforeEach(() => { + jest.isolateModules(() => { + jest.mock('ioredis', () => ({ + Redis: jest.fn().mockImplementation(() => ({ + on: jest.fn(), + quit: jest.fn().mockResolvedValue('OK'), + status: 'ready', + ping: jest.fn().mockResolvedValue('PONG'), + })), + })); + jest.mock('bullmq', () => ({ + Queue: jest.fn().mockImplementation((name: string) => ({ + name, + add: jest.fn().mockResolvedValue({ id: 'job-fence-1' }), + close: jest.fn().mockResolvedValue(undefined), + })), + QueueEvents: jest.fn().mockImplementation((name: string) => ({ + name, + on: jest.fn(), + })), + })); + const mod = require('../queues'); + (global as any).__queues = mod; + }); + }); + + test('nextFencingToken() increments monotonically', () => { + const { nextFencingToken } = (global as any).__queues; + expect(nextFencingToken('compound', 'vault-1')).toBe(1); + expect(nextFencingToken('compound', 'vault-1')).toBe(2); + expect(nextFencingToken('compound', 'vault-2')).toBe(1); // separate target + }); + + test('validateFencingToken() rejects stale tokens after new job enqueue', () => { + const { nextFencingToken, validateFencingToken } = (global as any).__queues; + nextFencingToken('compound', 'vault-1'); // 1 + nextFencingToken('compound', 'vault-1'); // 2 + + expect(validateFencingToken('compound', 'vault-1', 1)).toBe(false); + expect(validateFencingToken('compound', 'vault-1', 2)).toBe(true); + }); + + test('enqueueCompoundJob() attaches fencingToken and requiredSequence to payload', async () => { + const { enqueueCompoundJob } = (global as any).__queues; + const jobId = await enqueueCompoundJob('CVAULT_123', '1000000', 12345); + expect(jobId).toBeDefined(); + const { Queue } = require('bullmq'); + const instance = Queue.mock.results[0].value; + expect(instance.add).toHaveBeenCalledWith( + 'compound:CVAULT_123', + expect.objectContaining({ + vaultContractId: 'CVAULT_123', + fencingToken: 1, + requiredSequence: 12345, + }), + expect.any(Object), + ); + }); +}); + +describe('queues — retry budgets and poison isolation (#907)', () => { + beforeEach(() => { + jest.isolateModules(() => { + jest.mock('ioredis', () => ({ + Redis: jest.fn().mockImplementation(() => ({ + on: jest.fn(), + quit: jest.fn().mockResolvedValue('OK'), + status: 'ready', + ping: jest.fn().mockResolvedValue('PONG'), + })), + })); + jest.mock('bullmq', () => ({ + Queue: jest.fn().mockImplementation((name: string) => { + const instances = new Map(); + const self = { + name, + add: jest.fn().mockImplementation(async (jobName: string, data: any) => { + const id = `${name}:${jobName}:${Date.now()}`; + return { id, ...data }; + }), + getJobCounts: jest + .fn() + .mockImplementation((...keys: string[]) => { + if (name === 'poison') { + return { waiting: 2, active: 0, completed: 0, failed: 1, delayed: 0 }; + } + const counts: any = { waiting: 0, active: 0, completed: 0, failed: 0, delayed: 0 }; + keys.forEach((k: string) => { + if (k === 'failed') counts[k] = 12; + }); + return counts; + }), + close: jest.fn().mockResolvedValue(undefined), + opts: { connection: {} }, + }; + instances.set(name, self); + return self; + }), + QueueEvents: jest.fn().mockImplementation(() => ({ + on: jest.fn(), + })), + })); + const mod = require('../queues'); + (global as any).__queues = mod; + }); + }); + + test('classifyFailure() marks reverted simulation as non-retryable', () => { + const { classifyFailure } = (global as any).__queues; + const result = classifyFailure(new Error('Simulation failed: xyz')); + expect(result.retryable).toBe(false); + expect(result.reason).toBe('NON_RETRYABLE_ERROR'); + }); + + test('isRetryableError() distinguishes transient network blips from permanent failures', () => { + const { isRetryableError } = (global as any).__queues; + expect(isRetryableError('Network timeout')).toBe(true); + expect(isRetryableError('Contract reverted: harvest error')).toBe(false); + expect(isRetryableError('insufficient balance')).toBe(false); + }); + + test('getQueueHealth() reports warnings when failed jobs exceed threshold', async () => { + const { getQueueHealth, QUEUE_NAMES } = (global as any).__queues; + const mockQueue = { + name: QUEUE_NAMES.COMPOUND, + getJobCounts: jest.fn().mockResolvedValue({ + waiting: 0, + active: 1, + completed: 10, + failed: 12, + delayed: 0, + }), + opts: { connection: {} }, + } as any; + + const summary = await getQueueHealth([mockQueue]); + expect(summary.queues[0].warnings.some((w: string) => w.includes('failed'))).toBe(true); + }); +}); + describe('queues/types', () => { test('QUEUE_NAMES has LIQUIDATION and COMPOUND entries', () => { const { QUEUE_NAMES } = require('../queues/types'); diff --git a/backend/keepers/src/__tests__/redis.test.ts b/backend/keepers/src/__tests__/redis.test.ts index efd16c95a..10cf2eaa4 100644 --- a/backend/keepers/src/__tests__/redis.test.ts +++ b/backend/keepers/src/__tests__/redis.test.ts @@ -16,7 +16,8 @@ describe('redis utilities', () => { let _setRedisForTest: (r: import('ioredis').Redis) => void; const mockQuit = jest.fn().mockResolvedValue('OK'); const mockOn = jest.fn(); - const MockRedis = jest.fn().mockImplementation(() => ({ on: mockOn, quit: mockQuit, status: 'ready' })); + const mockPing = jest.fn().mockResolvedValue('PONG'); + const MockRedis = jest.fn().mockImplementation(() => ({ on: mockOn, quit: mockQuit, status: 'ready', ping: mockPing })); jest.isolateModules(() => { jest.mock('ioredis', () => ({ Redis: MockRedis })); @@ -26,7 +27,7 @@ describe('redis utilities', () => { _setRedisForTest = mod._setRedisForTest; }); - return { getRedis: getRedis!, closeRedis: closeRedis!, _setRedisForTest: _setRedisForTest!, MockRedis, mockQuit, mockOn }; + return { getRedis: getRedis!, closeRedis: closeRedis!, _setRedisForTest: _setRedisForTest!, MockRedis, mockQuit, mockOn, mockPing }; } // ── getRedis() ───────────────────────────────────────────────────────────── @@ -94,21 +95,82 @@ describe('redis utilities', () => { expect(MockRedis).not.toHaveBeenCalled(); }); - // #813: Add Redis reconnect and TTL expiry regression tests for keeper state - describe('Redis Reconnect and TTL Expiry (Keeper State)', () => { - test('covers reconnect scenarios after transient disconnection', () => { - const isReconnected = true; - expect(isReconnected).toBe(true); + // #909: Add Redis failover and lease recovery tests + describe('Redis Failover and Lease Recovery (#909)', () => { + test('detects degraded Redis via latency threshold', async () => { + const { getRedis, mockPing } = loadFresh(); + const redis = getRedis(); + mockPing.mockImplementation(() => new Promise((resolve) => setTimeout(() => resolve('PONG'), 1100))); + + const start = Date.now(); + await redis.ping(); + const latencyMs = Date.now() - start; + + expect(latencyMs).toBeGreaterThan(1000); + }); + + test('treats ping failure as outage, not degraded', async () => { + const { getRedis, mockPing } = loadFresh(); + const redis = getRedis(); + mockPing.mockRejectedValue(new Error('ECONNREFUSED')); + + await expect(redis.ping()).rejects.toThrow('ECONNREFUSED'); + }); + + test('connection events are registered for failover detection', () => { + const { getRedis, mockOn } = loadFresh(); + getRedis(); + const registeredEvents = mockOn.mock.calls.map(([event]) => event); + expect(registeredEvents).toContain('connect'); + expect(registeredEvents).toContain('error'); }); + }); - test('validates TTL-based cleanup of cached state', () => { - const stateExpired = true; - expect(stateExpired).toBe(true); + // #906: Stale lock fencing validation + describe('Stale Lock Fencing (#906)', () => { + test('fencing token changes block stale lock reuse', () => { + const { getRedis, _setRedisForTest } = loadFresh(); + const fakeRedis = { + on: jest.fn(), + quit: jest.fn().mockResolvedValue(undefined), + ping: jest.fn().mockResolvedValue('PONG'), + setex: jest.fn().mockResolvedValue('OK'), + get: jest.fn().mockResolvedValue(null), + } as any; + _setRedisForTest(fakeRedis); + + const redis = getRedis(); + // Simulate storing an old fencing record + fakeRedis.get = jest.fn().mockResolvedValue(JSON.stringify({ + jobId: 'job-old', + fencingToken: 5, + requiredSequence: 100, + updatedAt: new Date(Date.now() - 3600_000).toISOString(), + })); + + redis.get('keeper:job:attempt:liquidation:job-old').then((raw: string | null) => { + if (raw) { + const record = JSON.parse(raw); + expect(record.fencingToken).toBe(5); + expect(record.requiredSequence).toBe(100); + } + }); }); - test('ensures stale state does not revive unexpectedly', () => { - const staleStateRevived = false; - expect(staleStateRevived).toBe(false); + test('missing job record returns null (no stale state)', async () => { + const { getRedis, _setRedisForTest } = loadFresh(); + const fakeRedis = { + on: jest.fn(), + quit: jest.fn().mockResolvedValue(undefined), + ping: jest.fn().mockResolvedValue('PONG'), + setex: jest.fn().mockResolvedValue('OK'), + get: jest.fn().mockResolvedValue(null), + } as any; + _setRedisForTest(fakeRedis); + + const redis = getRedis(); + const raw = await redis.get('keeper:job:attempt:liquidation:missing'); + expect(raw).toBeNull(); }); }); -}); +}); \ No newline at end of file diff --git a/backend/keepers/src/api/queueHealth.ts b/backend/keepers/src/api/queueHealth.ts index b15ac4a6a..ebdb1285c 100644 --- a/backend/keepers/src/api/queueHealth.ts +++ b/backend/keepers/src/api/queueHealth.ts @@ -68,6 +68,24 @@ export function startKeeperHealthServer( return; } + if (req.method === 'GET' && url === '/health/redis') { + getQueueHealth(queues) + .then((summary) => { + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ + redisStatus: summary.redisStatus, + overallStatus: summary.overallStatus, + timestamp: summary.timestamp, + })); + }) + .catch((err: unknown) => { + logger.error({ err }, 'Redis health check failed'); + res.writeHead(503, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: 'Redis health check failed' })); + }); + return; + } + res.writeHead(404); res.end(); }); diff --git a/backend/keepers/src/queues/health.ts b/backend/keepers/src/queues/health.ts index 703d1673c..4ebac34ea 100644 --- a/backend/keepers/src/queues/health.ts +++ b/backend/keepers/src/queues/health.ts @@ -1,4 +1,6 @@ import { Queue } from 'bullmq'; +import { getRedisConnectionStatus } from './index'; +import { QUEUE_NAMES } from './types'; // --------------------------------------------------------------------------- // Types @@ -22,45 +24,25 @@ export interface QueueJobCounts { active: number; completed: number; failed: number; + delayed: number; + /** #907: Count of jobs quarantined to the poison queue */ poison: number; } -/** - * Metrics that go beyond simple counts — these help operators understand - * *how stale* the backlog is and *why* the most recent failure occurred. - */ -export interface QueueQualityMetrics { - /** - * Age in milliseconds of the oldest job currently in the `pending` state. - * `null` when there are no pending jobs. - */ - oldestPendingAgeMs: number | null; - /** - * The `failedReason` string from the most recently failed job. - * `null` when no failed jobs exist in the queue. - */ - latestFailureReason: string | null; -} - export interface QueueHealthEntry { name: string; counts: QueueJobCounts; - metrics: QueueQualityMetrics; - status: 'healthy' | 'warning'; + status: 'healthy' | 'degraded' | 'outage'; warnings: string[]; } export interface QueueHealthSummary { /** One entry per queue passed to `getQueueHealth`. */ queues: QueueHealthEntry[]; - /** - * Per-worker queue entries keyed by worker name. - * Provides the same structure as `queues` but scoped to named workers so - * dashboards can render compound vs. liquidation metrics side-by-side. - */ - workers: Record; - overallStatus: 'healthy' | 'warning'; + overallStatus: 'healthy' | 'degraded' | 'outage'; timestamp: string; + /** #909: Redis connection state derived from latency and errors */ + redisStatus: 'healthy' | 'degraded' | 'outage'; } // --------------------------------------------------------------------------- @@ -72,15 +54,8 @@ export const QUEUE_HEALTH_THRESHOLDS = { failed: Number(process.env.QUEUE_FAILED_THRESHOLD ?? '10'), /** Maximum number of delayed jobs before a `warning` is emitted. */ delayed: Number(process.env.QUEUE_DELAYED_THRESHOLD ?? '50'), - /** Maximum number of pending (waiting) jobs before a `warning` is emitted. */ - pending: Number(process.env.QUEUE_PENDING_THRESHOLD ?? '100'), - /** Maximum number of poison jobs before a `warning` is emitted. */ + /** #907: Retry budget exhaustion warning threshold */ poison: Number(process.env.QUEUE_POISON_THRESHOLD ?? '5'), - /** - * Maximum age (ms) of the oldest pending job before a `warning` is emitted. - * Default: 30 minutes. - */ - oldestPendingAgeMs: Number(process.env.QUEUE_OLDEST_PENDING_AGE_MS ?? String(30 * 60 * 1000)), } as const; // --------------------------------------------------------------------------- @@ -155,8 +130,7 @@ async function getLatestFailureReason(queue: Queue): Promise { * compound and liquidation workers explicitly. */ export async function getQueueHealth(queues: Queue[]): Promise { - const nowMs = Date.now(); - const t = QUEUE_HEALTH_THRESHOLDS; + const redisStatus = await getRedisConnectionStatus(); const entries = await Promise.all( queues.map(async (queue): Promise => { @@ -222,10 +196,24 @@ export async function getQueueHealth(queues: Queue[]): Promise QUEUE_HEALTH_THRESHOLDS.poison) { + warnings.push( + `poison jobs (${counts.poison}) exceed threshold (${QUEUE_HEALTH_THRESHOLDS.poison})`, + ); + } + + // #909: Degrade queue status if Redis is not healthy + let status: QueueHealthEntry['status'] = 'healthy'; + if (redisStatus === 'outage') { + status = 'outage'; + } else if (redisStatus === 'degraded' || warnings.length > 0) { + status = 'degraded'; + } return { name: queue.name, counts, + status, metrics, status: warnings.length > 0 ? 'warning' : 'healthy', warnings, @@ -233,6 +221,17 @@ export async function getQueueHealth(queues: Queue[]): Promise e.status === 'outage') + ? 'outage' + : entries.some((e) => e.status === 'degraded') + ? 'degraded' + : 'healthy'; + + return { + queues: entries, + overallStatus, // ── Per-worker map ───────────────────────────────────────────────────────── // Convention: "liquidation" → "liquidationWorker", "compound" → "compoundWorker" const workers: Record = {}; @@ -246,5 +245,6 @@ export async function getQueueHealth(queues: Queue[]): Promise e.status === 'warning') ? 'warning' : 'healthy', timestamp: new Date().toISOString(), + redisStatus, }; -} +} \ No newline at end of file diff --git a/backend/keepers/src/queues/index.ts b/backend/keepers/src/queues/index.ts index 93ae56e4c..1b67eabff 100644 --- a/backend/keepers/src/queues/index.ts +++ b/backend/keepers/src/queues/index.ts @@ -1,4 +1,4 @@ -import { Queue, QueueEvents } from 'bullmq'; +import { Queue, QueueEvents, Job } from 'bullmq'; import { getRedis } from '../utils/redis'; import { config } from '../config'; import { logger } from '../utils/logger'; @@ -6,8 +6,16 @@ import { QUEUE_NAMES, LiquidationJobData, CompoundJobData, + JOB_STATES, + TERMINAL_FAILURE_REASONS, + isRetryableError, + type JobAttemptRecord, + type QueueName, } from './types'; +export { isRetryableError, TERMINAL_FAILURE_REASONS, JOB_STATES, QUEUE_NAMES }; +export type { JobAttemptRecord, QueueName, LiquidationJobData, CompoundJobData }; + export { getQueueHealth } from './health'; export type { QueueHealthSummary, QueueHealthEntry, QueueJobCounts, QueueQualityMetrics } from './health'; @@ -18,6 +26,162 @@ const defaultJobOptions = { removeOnFail: { count: 500 }, }; +/** + * #907: Extended job options with retry budget control. + */ +export interface KeeperJobOptions { + attempts?: number; + backoff?: { type: 'exponential'; delay: number }; + removeOnComplete?: { count: number }; + removeOnFail?: { count: number }; +} + +/** + * #906: In-memory fence tracker for exactly-once execution. + * Maps (queueName, targetId) -> current fencing token expected by callers. + */ +const fenceStore = new Map(); + +/** + * #906: Atomically increment and return the next fencing token for a target. + * Callers must use this value when creating jobs; workers reject jobs whose + * token does not match the current store value. + */ +export function nextFencingToken(queueName: string, targetId: string): number { + const key = `${queueName}:${targetId}`; + const current = fenceStore.get(key) ?? 0; + const next = current + 1; + fenceStore.set(key, next); + logger.debug({ queueName, targetId, token: next }, 'Advanced fencing token'); + return next; +} + +/** + * #906: Validate that the job's fencing token matches the current store value. + * Returns true if the token is fresh; false if the job is stale. + */ +export function validateFencingToken(queueName: string, targetId: string, token: number): boolean { + const key = `${queueName}:${targetId}`; + const current = fenceStore.get(key) ?? 0; + if (token !== current) { + logger.warn( + { queueName, targetId, expected: current, actual: token }, + 'Fencing token mismatch — rejecting stale job', + ); + return false; + } + return true; +} + +/** + * #906: Persist a job attempt record for crash recovery and observability. + * Stored in a predictable Redis key so duplicate workers can detect claimed work. + */ +const JOB_RECORD_PREFIX = 'keeper:job:attempt:'; +const JOB_RECORD_TTL_MS = 24 * 60 * 60 * 1000; // 24 hours + +export async function persistJobRecord(record: JobAttemptRecord): Promise { + try { + const redis = getRedis(); + const key = `${JOB_RECORD_PREFIX}${record.queueName}:${record.jobId}`; + await redis.setex( + key, + Math.floor(JOB_RECORD_TTL_MS / 1000), + JSON.stringify(record), + ); + } catch (err) { + logger.warn({ err, jobId: record.jobId }, 'Failed to persist job attempt record'); + } +} + +/** + * #906: Fetch a persisted job attempt record. + */ +export async function getJobRecord( + queueName: QueueName, + jobId: string, +): Promise { + try { + const redis = getRedis(); + const raw = await redis.get(`${JOB_RECORD_PREFIX}${queueName}:${jobId}`); + if (!raw) return null; + return JSON.parse(raw) as JobAttemptRecord; + } catch { + return null; + } +} + +/** + * #907: Classify a job failure and return the terminal reason if non-retryable. + */ +export function classifyFailure(error: unknown): { + reason: string; + retryable: boolean; +} { + if (!error) { + return { reason: TERMINAL_FAILURE_REASONS.NON_RETRYABLE_ERROR, retryable: false }; + } + + const message = error instanceof Error ? error.message : String(error); + const retryable = isRetryableError(message); + + if (!retryable) { + let reason: string = TERMINAL_FAILURE_REASONS.NON_RETRYABLE_ERROR; + if (message.includes('SEQUENCE_MISMATCH') || message.includes('sequence')) { + reason = TERMINAL_FAILURE_REASONS.SEQUENCE_MISMATCH; + } else if (message.includes('FENCING_VIOLATION')) { + reason = TERMINAL_FAILURE_REASONS.FENCING_VIOLATION; + } else if (message.includes('QUOTA_EXCEEDED')) { + reason = TERMINAL_FAILURE_REASONS.QUOTA_EXCEEDED; + } + return { reason, retryable: false }; + } + + return { reason: 'RETRYABLE_ERROR', retryable: true }; +} + +/** + * #907: Move an exhausted job to the poison queue for operator review. + */ +export async function quarantineJob( + sourceQueue: Queue, + job: Job, + reason: string, + targetId: string, +): Promise { + try { + const poisonQueue = new Queue(QUEUE_NAMES.POISON, { connection: getRedis() }); + + // Preserve original payload plus failure metadata + await poisonQueue.add( + `poison:${job.id}`, + { + ...job.data, + _poisonReason: reason, + _originalJobId: job.id, + _quarantinedAt: new Date().toISOString(), + _targetId: targetId, + _failedAttempts: job.attemptsMade, + }, + { + attempts: 0, // No retries in poison + removeOnComplete: { count: 1000 }, + removeOnFail: { count: 1000 }, + }, + ); + + logger.warn( + { jobId: job.id, queueName: sourceQueue.name, reason, targetId }, + 'Job moved to poison queue', + ); + } catch (err) { + logger.error( + { err, jobId: job.id, queueName: sourceQueue.name }, + 'Failed to quarantine job', + ); + } +} + /** * BullMQ Queue for liquidation jobs. * Each job carries the account address and position snapshot that triggered it. @@ -39,6 +203,119 @@ export function createCompoundQueue(): Queue { }); } +/** + * #906: Enqueue a liquidation job with sequence and fence metadata. + */ +export async function enqueueLiquidationJob( + accountAddress: string, + currentCrBps: number, + collateralValueUsd: string, + debtAmount: string, + requiredSequence: number, +): Promise { + const queue = createLiquidationQueue(); + const fencingToken = nextFencingToken(QUEUE_NAMES.LIQUIDATION, accountAddress); + + const job = await queue.add( + `liquidation:${accountAddress}`, + { + accountAddress, + currentCrBps, + collateralValueUsd, + debtAmount, + fencingToken, + requiredSequence, + } as LiquidationJobData, + { jobId: `${Date.now()}-${Math.random().toString(36).slice(2, 8)}` }, + ); + + await persistJobRecord({ + jobId: job.id!, + queueName: QUEUE_NAMES.LIQUIDATION, + state: JOB_STATES.CREATED, + attemptNumber: 0, + fencingToken, + requiredSequence, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: accountAddress, + }); + + logger.info( + { jobId: job.id, accountAddress, fencingToken, requiredSequence }, + 'Liquidation job enqueued', + ); + + return job.id!; +} + +/** + * #906: Enqueue a compound job with sequence and fence metadata. + */ +export async function enqueueCompoundJob( + vaultContractId: string, + minHarvestAmount: string, + requiredSequence: number, +): Promise { + const queue = createCompoundQueue(); + const fencingToken = nextFencingToken(QUEUE_NAMES.COMPOUND, vaultContractId); + + const job = await queue.add( + `compound:${vaultContractId}`, + { + vaultContractId, + minHarvestAmount, + fencingToken, + requiredSequence, + } as CompoundJobData, + { jobId: `${Date.now()}-${Math.random().toString(36).slice(2, 8)}` }, + ); + + await persistJobRecord({ + jobId: job.id!, + queueName: QUEUE_NAMES.COMPOUND, + state: JOB_STATES.CREATED, + attemptNumber: 0, + fencingToken, + requiredSequence, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: vaultContractId, + }); + + logger.info( + { jobId: job.id, vaultContractId, fencingToken, requiredSequence }, + 'Compound job enqueued', + ); + + return job.id!; +} + +/** + * #909: Get degraded Redis connection status for health checks. + * Distinguishes between degraded (slow/flaky) and total outage. + */ +export async function getRedisConnectionStatus(): Promise< + 'healthy' | 'degraded' | 'outage' +> { + try { + const redis = getRedis(); + const start = Date.now(); + await redis.ping(); + const latencyMs = Date.now() - start; + + if (latencyMs > 1000) { + logger.warn({ latencyMs }, 'Redis degraded — high latency'); + return 'degraded'; + } + + return 'healthy'; + } catch (err) { + logger.error({ err }, 'Redis connection outage'); + return 'outage'; + } +} + /** * Attach event listeners that log queue lifecycle events for observability. * Call this once per queue after creation. @@ -46,14 +323,14 @@ export function createCompoundQueue(): Queue { export function attachQueueEvents(queueName: string): QueueEvents { const events = new QueueEvents(queueName, { connection: getRedis() }); - events.on('completed', ({ jobId }) => - logger.info({ queueName, jobId }, 'Job completed'), + events.on('completed', (args: { jobId: string }) => + logger.info({ queueName, jobId: args.jobId }, 'Job completed'), ); - events.on('failed', ({ jobId, failedReason }) => - logger.error({ queueName, jobId, failedReason }, 'Job failed'), + events.on('failed', (args: { jobId: string; failedReason: string }) => + logger.error({ queueName, jobId: args.jobId, failedReason: args.failedReason }, 'Job failed'), ); - events.on('stalled', ({ jobId }) => - logger.warn({ queueName, jobId }, 'Job stalled'), + events.on('stalled', (args: { jobId: string }) => + logger.warn({ queueName, jobId: args.jobId }, 'Job stalled'), ); return events; diff --git a/backend/keepers/src/queues/types.ts b/backend/keepers/src/queues/types.ts index 362af1873..b384e6a6a 100644 --- a/backend/keepers/src/queues/types.ts +++ b/backend/keepers/src/queues/types.ts @@ -5,6 +5,8 @@ export const QUEUE_NAMES = { LIQUIDATION: 'liquidation', COMPOUND: 'compound', + /** Jobs moved after exceeding retry budget (terminal failure) */ + POISON: 'poison', } as const; export type QueueName = (typeof QUEUE_NAMES)[keyof typeof QUEUE_NAMES]; @@ -19,6 +21,10 @@ export interface LiquidationJobData { collateralValueUsd: string; /** Outstanding debt in sUSD */ debtAmount: string; + /** #906: Fencing token incremented on each attempt to reject stale locks */ + fencingToken: number; + /** #906: Sequence number required before submission */ + requiredSequence: number; /** * Unix ms timestamp of the oracle price used to compute `currentCrBps`. * Omitted when the scanner didn't attach price provenance (dry-run policy @@ -39,4 +45,78 @@ export interface CompoundJobData { vaultContractId: string; /** Expected minimum harvest amount (slippage guard) */ minHarvestAmount: string; + /** #906: Fencing token incremented on each attempt to reject stale locks */ + fencingToken: number; + /** #906: Sequence number required before submission */ + requiredSequence: number; } + +/** + * #907: Terminal failure reasons that exhaust the retry budget. + * Once exhausted, the job is moved to the POISON queue. + */ +export const TERMINAL_FAILURE_REASONS = { + NON_RETRYABLE_ERROR: 'NON_RETRYABLE_ERROR', + SEQUENCE_MISMATCH: 'SEQUENCE_MISMATCH', + FENCING_VIOLATION: 'FENCING_VIOLATION', + RETRY_BUDGET_EXHAUSTED: 'RETRY_BUDGET_EXHAUSTED', + QUOTA_EXCEEDED: 'QUOTA_EXCEEDED', +} as const; + +export type TerminalFailureReason = + (typeof TERMINAL_FAILURE_REASONS)[keyof typeof TERMINAL_FAILURE_REASONS]; + +/** + * #907: Retry classification for errors. + * Retryable errors indicate transient conditions (network blips, temporary congestion). + * Non-retryable errors indicate permanent failures that should not be retried. + */ +export function isRetryableError(error: Error | string): boolean { + const message = typeof error === 'string' ? error : error.message; + const nonRetryablePatterns = [ + 'Simulation failed', + 'Contract reverted', + 'NON_RETRYABLE_ERROR', + 'FENCING_VIOLATION', + 'QUOTA_EXCEEDED', + 'insufficient balance', + 'Auth required', + 'Permission denied', + ]; + + return !nonRetryablePatterns.some((pattern) => + message.toLowerCase().includes(pattern.toLowerCase()), + ); +} + +/** + * #906: Job lifecycle states for observability and crash recovery. + */ +export const JOB_STATES = { + CREATED: 'created', + CLAIMED: 'claimed', // Worker acquired lock + SUBMITTED: 'submitted', // Transaction sent + CONFIRMED: 'confirmed', // On-chain success + FAILED: 'failed', // Transient failure (will retry) + EXHAUSTED: 'exhausted', // Terminal failure (moved to poison) +} as const; + +export type JobState = (typeof JOB_STATES)[keyof typeof JOB_STATES]; + +/** + * #906: Persisted job attempt record for crash recovery and exactly-once semantics. + */ +export interface JobAttemptRecord { + jobId: string; + queueName: QueueName; + state: JobState; + attemptNumber: number; + fencingToken: number; + requiredSequence: number; + txHash?: string; + failedReason?: string; + claimedAt: string; + updatedAt: string; + /** Vault or account identifier for logging */ + targetId: string; +} \ No newline at end of file diff --git a/backend/keepers/src/signer/KeeperSigner.ts b/backend/keepers/src/signer/KeeperSigner.ts index 461f4d8fe..ee3e42ad4 100644 --- a/backend/keepers/src/signer/KeeperSigner.ts +++ b/backend/keepers/src/signer/KeeperSigner.ts @@ -122,6 +122,7 @@ export class KeeperSigner { contractId: string, method: string, args: xdr.ScVal[] = [], + options?: { requiredSequence?: number; fencingToken?: number; persistRecord?: (record: import('../queues/types').JobAttemptRecord) => Promise }, auditContext?: KeeperAuditContext, ): Promise { if (!this.provider.allowedOperations.has(method)) { @@ -135,6 +136,15 @@ export class KeeperSigner { const account = await this.server.getAccount(this.provider.publicKey); + if (options?.requiredSequence !== undefined) { + const currentSequence = Number((account as any).sequence); + if (currentSequence !== options.requiredSequence) { + throw new Error( + `SEQUENCE_MISMATCH: expected sequence ${options.requiredSequence}, got ${currentSequence}`, + ); + } + } + const contract = new Contract(contractId); const op = contract.call(method, ...args); @@ -185,6 +195,21 @@ export class KeeperSigner { const hash = sendResult.hash; logger.info({ hash, method, contractId }, 'Transaction submitted'); + if (options?.persistRecord && options.fencingToken !== undefined) { + await options.persistRecord({ + jobId: hash, + queueName: 'compound', + state: 'submitted', + attemptNumber: 0, + fencingToken: options.fencingToken, + requiredSequence: options.requiredSequence ?? 0, + txHash: hash, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: contractId, + }); + } + try { await this.pollForConfirmation(hash); } catch (err) { diff --git a/backend/keepers/src/workers/CompoundWorker.ts b/backend/keepers/src/workers/CompoundWorker.ts index 70ebd6176..2bc240df2 100644 --- a/backend/keepers/src/workers/CompoundWorker.ts +++ b/backend/keepers/src/workers/CompoundWorker.ts @@ -4,7 +4,15 @@ import { getRedis } from '../utils/redis'; import { config } from '../config'; import { logger } from '../utils/logger'; import { KeeperSigner } from '../signer/KeeperSigner'; -import { QUEUE_NAMES, CompoundJobData } from '../queues/types'; +import { QUEUE_NAMES, CompoundJobData, JOB_STATES } from '../queues/types'; +import { + validateFencingToken, + getJobRecord, + persistJobRecord, + classifyFailure, + quarantineJob, + createCompoundQueue, +} from '../queues'; /** * CompoundWorker processes auto-compound jobs. @@ -17,6 +25,15 @@ import { QUEUE_NAMES, CompoundJobData } from '../queues/types'; * * Jobs are produced by the CompoundScheduler on a time-based schedule and * can also be triggered manually via the admin API. + * + * #906 Exactly-once guarantees: + * - Workers validate the fencing token before execution. + * - Job attempt records are persisted to Redis for crash recovery. + * - Required Stellar sequence is verified before submission. + * + * #907 Poison isolation: + * - Non-retryable failures are classified and moved to the poison queue. + * - Retryable failures are re-queued by BullMQ with exponential backoff. */ export class CompoundWorker { private readonly worker: Worker; @@ -48,23 +65,88 @@ export class CompoundWorker { * @param job - BullMQ Job containing CompoundJobData */ async process(job: Job): Promise<{ txHash: string }> { - const { vaultContractId, minHarvestAmount } = job.data; + const { vaultContractId, minHarvestAmount, fencingToken, requiredSequence } = job.data; logger.info( - { jobId: job.id, vaultContractId }, + { jobId: job.id, vaultContractId, fencingToken, requiredSequence }, '[CompoundWorker] Processing compound job', ); + // #906: Reject stale jobs whose fencing token no longer matches + if (!validateFencingToken(QUEUE_NAMES.COMPOUND, vaultContractId, fencingToken)) { + throw new Error(`FENCING_VIOLATION: stale fencing token for vault ${vaultContractId}`); + } + + // Mark claimed and persist attempt record + try { + await persistJobRecord({ + jobId: job.id!, + queueName: QUEUE_NAMES.COMPOUND, + state: JOB_STATES.CLAIMED, + attemptNumber: job.attemptsMade, + fencingToken, + requiredSequence, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: vaultContractId, + }); + } catch (err) { + logger.warn({ err, jobId: job.id }, 'Failed to persist claim record'); + } + + // #906: Fetch current Stellar sequence and verify it matches the job requirement + try { + const account = await this.signer['server'].getAccount(this.signer.publicKey); + const currentSequence = parseInt((account as any).sequence, 10); + if (currentSequence !== requiredSequence) { + throw new Error( + `SEQUENCE_MISMATCH: expected sequence ${requiredSequence}, got ${currentSequence}`, + ); + } + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + const { retryable } = classifyFailure(err); + if (!retryable) { + await quarantineJob(createCompoundQueue(), job, message, vaultContractId); + } + throw err; + } + // harvest(caller: Address, min_amount: i128) const keeperScVal = new Address(this.signer.publicKey).toScVal(); const minAmtXdr = nativeToScVal(BigInt(minHarvestAmount), { type: 'i128' }); - const txHash = await this.signer.invokeContract( - vaultContractId, - 'harvest', - [keeperScVal, minAmtXdr], - { workerName: 'CompoundWorker', jobId: job.id, policyVersion: 'v1' }, - ); + let txHash: string; + try { + txHash = await this.signer.invokeContract( + vaultContractId, + 'harvest', + [keeperScVal, minAmtXdr], + undefined, + { workerName: 'CompoundWorker', jobId: job.id, policyVersion: 'v1' }, + ); + + // Persist submitted state + await persistJobRecord({ + jobId: job.id!, + queueName: QUEUE_NAMES.COMPOUND, + state: JOB_STATES.SUBMITTED, + attemptNumber: job.attemptsMade, + fencingToken, + requiredSequence, + txHash, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: vaultContractId, + }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + const { retryable, reason } = classifyFailure(err); + if (!retryable) { + await quarantineJob(createCompoundQueue(), job, reason, vaultContractId); + } + throw err; + } logger.info( { jobId: job.id, vaultContractId, txHash }, @@ -79,4 +161,4 @@ export class CompoundWorker { await this.worker.close(); logger.info('[CompoundWorker] Worker closed'); } -} +} \ No newline at end of file diff --git a/backend/keepers/src/workers/LiquidationWorker.ts b/backend/keepers/src/workers/LiquidationWorker.ts index 7608abe6d..726e59a0e 100644 --- a/backend/keepers/src/workers/LiquidationWorker.ts +++ b/backend/keepers/src/workers/LiquidationWorker.ts @@ -4,7 +4,15 @@ import { getRedis } from '../utils/redis'; import { config } from '../config'; import { logger } from '../utils/logger'; import { KeeperSigner } from '../signer/KeeperSigner'; -import { QUEUE_NAMES, LiquidationJobData } from '../queues/types'; +import { QUEUE_NAMES, LiquidationJobData, JOB_STATES } from '../queues/types'; +import { + validateFencingToken, + getJobRecord, + persistJobRecord, + classifyFailure, + quarantineJob, + createLiquidationQueue, +} from '../queues'; /** Maximum age of the oracle price backing `currentCrBps` before it's considered stale. */ export const MAX_PRICE_AGE_MS = 5 * 60 * 1000; // 5 minutes @@ -96,13 +104,51 @@ export class LiquidationWorker { * @param job - BullMQ Job containing LiquidationJobData */ async process(job: Job): Promise<{ txHash: string }> { - const { accountAddress } = job.data; + const { accountAddress, fencingToken, requiredSequence } = job.data; logger.info( - { jobId: job.id, accountAddress, crBps: job.data.currentCrBps }, + { jobId: job.id, accountAddress, crBps: job.data.currentCrBps, fencingToken, requiredSequence }, '[LiquidationWorker] Processing liquidation job', ); + // #906: Reject stale jobs whose fencing token no longer matches + if (!validateFencingToken(QUEUE_NAMES.LIQUIDATION, accountAddress, fencingToken)) { + throw new Error(`FENCING_VIOLATION: stale fencing token for account ${accountAddress}`); + } + + // Mark claimed and persist attempt record + try { + await persistJobRecord({ + jobId: job.id!, + queueName: QUEUE_NAMES.LIQUIDATION, + state: JOB_STATES.CLAIMED, + attemptNumber: job.attemptsMade, + fencingToken, + requiredSequence, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: accountAddress, + }); + } catch (err) { + logger.warn({ err, jobId: job.id }, 'Failed to persist claim record'); + } + + // #906: Fetch current Stellar sequence and verify it matches the job requirement + try { + const account = await this.signer['server'].getAccount(this.signer.publicKey); + const currentSequence = parseInt((account as any).sequence, 10); + if (currentSequence !== requiredSequence) { + throw new Error( + `SEQUENCE_MISMATCH: expected sequence ${requiredSequence}, got ${currentSequence}`, + ); + } + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + const { retryable } = classifyFailure(err); + if (!retryable) { + await quarantineJob(createLiquidationQueue(), job, message, accountAddress); + } + throw err; const dryRun = evaluateLiquidationDryRun(job.data); if (!dryRun.safe) { logger.warn( @@ -116,12 +162,36 @@ export class LiquidationWorker { const liquidatorScVal = new Address(this.signer.publicKey).toScVal(); const userScVal = new Address(accountAddress).toScVal(); - const txHash = await this.signer.invokeContract( - config.contracts.stablecoinManager, - 'liquidate', - [liquidatorScVal, userScVal], - { workerName: 'LiquidationWorker', jobId: job.id, policyVersion: 'v1' }, - ); + let txHash: string; + try { + txHash = await this.signer.invokeContract( + config.contracts.stablecoinManager, + 'liquidate', + [liquidatorScVal, userScVal], + undefined, + { workerName: 'LiquidationWorker', jobId: job.id, policyVersion: 'v1' }, + ); + + await persistJobRecord({ + jobId: job.id!, + queueName: QUEUE_NAMES.LIQUIDATION, + state: JOB_STATES.SUBMITTED, + attemptNumber: job.attemptsMade, + fencingToken, + requiredSequence, + txHash, + claimedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + targetId: accountAddress, + }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + const { retryable, reason } = classifyFailure(err); + if (!retryable) { + await quarantineJob(createLiquidationQueue(), job, reason, accountAddress); + } + throw err; + } logger.info( { jobId: job.id, accountAddress, txHash }, @@ -136,4 +206,4 @@ export class LiquidationWorker { await this.worker.close(); logger.info('[LiquidationWorker] Worker closed'); } -} +} \ No newline at end of file diff --git a/server/src/services/__tests__/pnlCalculator.test.ts b/server/src/services/__tests__/pnlCalculator.test.ts new file mode 100644 index 000000000..a5eba989a --- /dev/null +++ b/server/src/services/__tests__/pnlCalculator.test.ts @@ -0,0 +1,124 @@ +import { calculatePnL, calculateTWR } from '../pnl_engine/pnlCalculator'; +import type { UserTransaction, SharePriceSnapshot } from '../pnl_engine/pnlCalculator'; + +function makeDate(iso: string): Date { + return new Date(iso); +} + +describe('PnL Engine — Historical Valuation Cache', () => { + // #894: Reject stale valuations from PnL by default + describe('stale price quarantine', () => { + const transactions: UserTransaction[] = [ + { + action: 'DEPOSIT', + amount: 1000, + shares: 100, + sharePriceAtTx: 10, + timestamp: makeDate('2024-01-01'), + }, + ]; + + const stalePrice = makeDate('2023-12-01'); // 1 month stale + const freshPrice = makeDate('2024-01-02'); // fresh + + it('refuses a stale valuation unless explicitly requested', () => { + const staleHistory: SharePriceSnapshot[] = [ + { snapshotAt: stalePrice, sharePrice: 10.5 }, + ]; + + // Default behavior: reject stale data by requiring non-empty fresh history + const result = calculatePnL(transactions, staleHistory, 10.5); + + // With only stale data and no override, PnL should not crash but should be conservative + expect(result.totalDeposited).toBe(1000); + expect(result.absolutePnL).toBeCloseTo(0, 1); + }); + + it('uses mixed history when at least one fresh snapshot exists', () => { + const mixedHistory: SharePriceSnapshot[] = [ + { snapshotAt: stalePrice, sharePrice: 10.5 }, + { snapshotAt: freshPrice, sharePrice: 11.0 }, + ]; + + const result = calculatePnL(transactions, mixedHistory, 11.0); + expect(result.dailySnapshots.length).toBeGreaterThan(0); + expect(result.dailySnapshots[result.dailySnapshots.length - 1].sharePrice).toBeCloseTo(11.0, 1); + }); + + it('does not produce misleading PnL from a missing source alone', () => { + const emptyHistory: SharePriceSnapshot[] = []; + const result = calculatePnL(transactions, emptyHistory, 10); + + expect(result.dailySnapshots).toEqual([]); + expect(result.absolutePnL).toBeCloseTo(0, 1); + }); + }); + + // #894: Conflicting price sources + describe('conflicting source handling', () => { + it('reconciles overlapping snapshots by latest timestamp', () => { + const transactions: UserTransaction[] = [ + { + action: 'DEPOSIT', + amount: 500, + shares: 50, + sharePriceAtTx: 10, + timestamp: makeDate('2024-02-01'), + }, + ]; + + const history: SharePriceSnapshot[] = [ + { snapshotAt: makeDate('2024-02-01'), sharePrice: 10.0 }, + { snapshotAt: makeDate('2024-02-01T12:00:00'), sharePrice: 10.2 }, + { snapshotAt: makeDate('2024-02-02'), sharePrice: 10.4 }, + ]; + + const result = calculatePnL(transactions, history, 10.4); + expect(result.dailySnapshots.length).toBeGreaterThan(0); + expect(result.dailySnapshots[0].sharePrice).toBeCloseTo(10.2, 1); + }); + }); + + // #894: Missing price handling + describe('missing price handling', () => { + it('falls back to tx price when no history is available', () => { + const transactions: UserTransaction[] = [ + { + action: 'DEPOSIT', + amount: 200, + shares: 20, + sharePriceAtTx: 12.0, + timestamp: makeDate('2024-03-01'), + }, + { + action: 'WITHDRAW', + amount: 100, + shares: -8, + sharePriceAtTx: 12.5, + timestamp: makeDate('2024-03-15'), + }, + ]; + + const result = calculatePnL(transactions, [], 12.5); + expect(result.currentValue).toBeCloseTo(120, 1); + expect(result.absolutePnL).toBeCloseTo(0, 1); + }); + }); + + // Standard PnL math sanity checks + describe('core PnL math', () => { + it('computes TWR correctly across deposits and withdrawals', () => { + const txs: UserTransaction[] = [ + { action: 'DEPOSIT', amount: 1000, shares: 100, sharePriceAtTx: 10, timestamp: makeDate('2024-01-01') }, + { action: 'WITHDRAW', amount: 200, shares: -20, sharePriceAtTx: 12, timestamp: makeDate('2024-02-01') }, + ]; + const prices: SharePriceSnapshot[] = [ + { snapshotAt: makeDate('2024-01-01'), sharePrice: 10 }, + { snapshotAt: makeDate('2024-02-01'), sharePrice: 12 }, + { snapshotAt: makeDate('2024-03-01'), sharePrice: 15 }, + ]; + + expect(calculateTWR(txs, prices, 15)).toBeCloseTo(0.25, 2); + }); + }); +}); \ No newline at end of file diff --git a/server/src/services/__tests__/treasurySimulation.test.ts b/server/src/services/__tests__/treasurySimulation.test.ts new file mode 100644 index 000000000..b5b287b43 --- /dev/null +++ b/server/src/services/__tests__/treasurySimulation.test.ts @@ -0,0 +1,67 @@ +import { simulateTreasury, assertValidScenarioInput } from '../treasurySimulationService'; + +describe('Treasury Simulation Service — Valuation Provenance (#894)', () => { + it('accepts scenarios with explicit provenance fields', () => { + const input = { + id: 'ts-1', + name: 'Conservative', + totalCapitalUsd: 1_000_000, + allocations: [ + { + vaultId: 'V1', + vaultName: 'StableVault', + allocationPct: 60, + apy: 4.5, + tvlUsd: 10_000_000, + riskScore: 2, + rotationCostPct: 0.1, + priceSource: 'chainlink', + priceTimestamp: new Date().toISOString(), + confidence: 0.95, + }, + { + vaultId: 'V2', + vaultName: 'GrowthVault', + allocationPct: 40, + apy: 8.2, + tvlUsd: 5_000_000, + riskScore: 6, + rotationCostPct: 0.5, + priceSource: 'direct-feed', + priceTimestamp: new Date(Date.now() - 120_000).toISOString(), + confidence: 0.8, + }, + ], + createdAt: new Date().toISOString(), + }; + + expect(() => assertValidScenarioInput(input)).not.toThrow(); + const scenario = assertValidScenarioInput(input); + const result = simulateTreasury(scenario); + expect(result.projectedYieldUsd).toBeGreaterThan(0); + expect(result.allocationBreakdown).toHaveLength(2); + }); + + it('simulateTreasury uses APY inputs and documents.source fields', () => { + const scenario = assertValidScenarioInput({ + id: 'ts-2', + name: 'APY-check', + totalCapitalUsd: 500_000, + allocations: [ + { + vaultId: 'V3', + vaultName: 'BlueChip', + allocationPct: 100, + apy: 6.0, + tvlUsd: 50_000_000, + riskScore: 3, + rotationCostPct: 0.2, + }, + ], + }); + + const result = simulateTreasury(scenario); + expect(result.projectedYieldPct).toBeCloseTo(6.0, 1); + expect(result.allocationBreakdown[0].projectedYieldUsd).toBeCloseTo(30_000, 0); + }); +}); \ No newline at end of file diff --git a/todo-app/index.html b/todo-app/index.html new file mode 100644 index 000000000..6f7c7a303 --- /dev/null +++ b/todo-app/index.html @@ -0,0 +1,641 @@ + + + + + + Todo App + + + +
+ +
+

My Todos

+
+ + +
+ + +
+ + +
+ + + +
+ + +
    + + +
    + 0 items left + +
    +
    + + + + \ No newline at end of file