- Note: Data synchronization is handled automatically by scheduled tasks that run every six hours. A score lag of up to 1 epoch can be expected between sync cycles. + Note: Data synchronization is handled automatically by scheduled tasks that run every hour. A score lag of up to 1 epoch can be expected between sync cycles.
diff --git a/nuxt.config.ts b/nuxt.config.ts index bc44d2d..595ea53 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -162,8 +162,8 @@ export default defineNuxtConfig({ tasks: true, }, scheduledTasks: { - // Six-hour sync: wrapper task records run + executes sync tasks - '0 */6 * * *': ['cron:sync'], + // Hourly sync: wrapper task records run + executes sync tasks + '0 * * * *': ['cron:sync'], }, openAPI: { meta: { title: name, description, version }, diff --git a/package.json b/package.json index 600165c..5aca0c6 100644 --- a/package.json +++ b/package.json @@ -17,7 +17,7 @@ "generate": "nuxt generate", "worker:dev": "npx wrangler --cwd .output dev", "worker:dev:mainnet:prod": "tsx scripts/dev-mainnet-prod-worker.ts", - "worker:trigger-sync": "curl 'http://localhost:8787/__scheduled?cron=0+*/6+*+*+*'", + "worker:trigger-sync": "curl 'http://localhost:8787/__scheduled?cron=0+*+*+*+*'", "postinstall": "nuxt prepare", "prepublishOnly": "nr build", "typecheck": "nuxt typecheck && nr -r typecheck", diff --git a/server/tasks/cron/sync.test.ts b/server/tasks/cron/sync.test.ts index 3757945..9436d5b 100644 --- a/server/tasks/cron/sync.test.ts +++ b/server/tasks/cron/sync.test.ts @@ -69,11 +69,11 @@ afterAll(() => { }) describe('scheduled synchronization order', () => { - it('records the six-hour schedule', async () => { + it('records the hourly schedule', async () => { await task.run({ payload: {}, context: {} } as never) expect(mocks.insertValues).toHaveBeenCalledWith(expect.objectContaining({ - cron: '0 */6 * * *', + cron: '0 * * * *', })) }) diff --git a/server/tasks/cron/sync.ts b/server/tasks/cron/sync.ts index bc30226..1629154 100644 --- a/server/tasks/cron/sync.ts +++ b/server/tasks/cron/sync.ts @@ -4,7 +4,7 @@ import { runTask } from 'nitropack/runtime' import { runTasksBestEffort } from '~~/server/utils/cron-task-runner' import { eq, tables, useDrizzle } from '~~/server/utils/drizzle' -const CRON_EXPRESSION = '0 */6 * * *' +const CRON_EXPRESSION = '0 * * * *' const TASKS: string[] = ['sync:epochs', 'sync:snapshot', 'sync:scores'] interface FailedTask { diff --git a/server/tasks/sync/epochs.test.ts b/server/tasks/sync/epochs.test.ts index 8d82556..cefbfde 100644 --- a/server/tasks/sync/epochs.test.ts +++ b/server/tasks/sync/epochs.test.ts @@ -103,7 +103,7 @@ beforeEach(async () => { mocks.synchronizeCompletedEpoch .mockResolvedValueOnce({ epochNumber: 100, status: 'failed', error: 'first RPC failure', repairAttempted: true }) .mockResolvedValueOnce({ epochNumber: 99, status: 'verified' }) - .mockResolvedValueOnce({ epochNumber: 98, status: 'failed', error: 'repair budget exhausted', repairAttempted: false }) + .mockResolvedValueOnce({ epochNumber: 98, status: 'verified' }) .mockResolvedValueOnce({ epochNumber: 97, status: 'verified' }) .mockResolvedValueOnce({ epochNumber: 96, status: 'already_finalized' }) mocks.sendSyncFailureNotification.mockResolvedValue(undefined) @@ -151,7 +151,8 @@ describe('planned completed-epoch synchronization', () => { expect(isFullRepairAllowed(0, true)).toBe(true) expect(isFullRepairAllowed(5, true)).toBe(true) expect(isFullRepairAllowed(0, false)).toBe(true) - expect(isFullRepairAllowed(1, false)).toBe(false) + expect(isFullRepairAllowed(1, false)).toBe(true) + expect(isFullRepairAllowed(4, false)).toBe(false) }) it('limits production repairs while returning the failed epoch details', async () => { @@ -173,27 +174,83 @@ describe('planned completed-epoch synchronization', () => { limit: 50, syncingLeaseDurationMs: 6 * 60 * 60 * 1000, })) - expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(1, 100, 'testnet', { allowRepair: true }) - expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(2, 99, 'testnet', { allowRepair: false }) - expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(3, 98, 'testnet', { allowRepair: false }) - expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(4, 97, 'testnet', { allowRepair: false }) - expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(5, 96, 'testnet', { allowRepair: false }) + expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(1, 100, 'testnet') + expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(2, 99, 'testnet') + expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(3, 98, 'testnet') + expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(4, 97, 'testnet') + expect(mocks.synchronizeCompletedEpoch).toHaveBeenNthCalledWith(5, 96, 'testnet') expect(mocks.synchronizeCompletedEpoch).toHaveBeenCalledTimes(5) expect(result).toEqual({ result: { success: false, error: '[sync:epochs] epoch 100 failed: first RPC failure', - totalSynced: 2, - epochsSynced: [99, 97], + totalSynced: 3, + epochsSynced: [99, 98, 97], + deferredEpochs: [], outcomes: [ { epochNumber: 100, status: 'failed', error: 'first RPC failure', repairAttempted: true }, { epochNumber: 99, status: 'verified' }, - { epochNumber: 98, status: 'failed', error: 'repair budget exhausted', repairAttempted: false }, + { epochNumber: 98, status: 'verified' }, { epochNumber: 97, status: 'verified' }, { epochNumber: 96, status: 'already_finalized' }, ], }, }) }) + it('repairs a bounded batch and leaves the remaining epochs untouched for the next run', async () => { + mocks.planEpochSync.mockReturnValue([100, 99, 98, 97, 96, 95]) + mocks.synchronizeCompletedEpoch.mockReset().mockImplementation(async (epochNumber: number) => ({ + epochNumber, + status: 'repaired', + })) + + const { result } = await task.run() + + expect(result).toMatchObject({ + success: true, + totalSynced: 4, + epochsSynced: [100, 99, 98, 97], + deferredEpochs: [96, 95], + }) + expect(mocks.synchronizeCompletedEpoch.mock.calls.map(([epoch]) => epoch)).toEqual([100, 99, 98, 97]) + expect(mocks.sendSyncFailureNotification).not.toHaveBeenCalled() + }) + + it('counts failed repair attempts toward the batch limit and retains real failures', async () => { + mocks.synchronizeCompletedEpoch.mockReset().mockImplementation(async (epochNumber: number) => ({ + epochNumber, + status: 'failed', + error: 'RPC unavailable', + repairAttempted: true, + })) + + const { result } = await task.run() + + expect(result).toMatchObject({ success: false, totalSynced: 0, deferredEpochs: [96] }) + expect(mocks.synchronizeCompletedEpoch).toHaveBeenCalledTimes(4) + expect(mocks.sendSyncFailureNotification).toHaveBeenCalledWith('missing-epoch', expect.objectContaining({ + message: expect.stringContaining('RPC unavailable'), + })) + }) + + it('stops starting epochs after the runtime budget while preserving completed progress', async () => { + vi.useFakeTimers() + try { + vi.setSystemTime(new Date('2026-09-24T10:00:00Z')) + mocks.synchronizeCompletedEpoch.mockReset().mockImplementation(async (epochNumber: number) => { + vi.setSystemTime(new Date('2026-09-24T10:08:00Z')) + return { epochNumber, status: 'repaired' } + }) + + const { result } = await task.run() + + expect(result).toMatchObject({ success: true, epochsSynced: [100], deferredEpochs: [99, 98, 97, 96] }) + expect(mocks.synchronizeCompletedEpoch).toHaveBeenCalledTimes(1) + expect(mocks.sendSyncFailureNotification).not.toHaveBeenCalled() + } + finally { + vi.useRealTimers() + } + }) }) diff --git a/server/tasks/sync/epochs.ts b/server/tasks/sync/epochs.ts index ceed21b..bcc59e4 100644 --- a/server/tasks/sync/epochs.ts +++ b/server/tasks/sync/epochs.ts @@ -14,6 +14,8 @@ import { getOldestV2BackfillEpoch } from '~~/server/utils/scores' import { sendSyncFailureNotification } from '~~/server/utils/slack' const MAX_PRODUCTION_EPOCH_CANDIDATES_PER_RUN = 50 +const MAX_PRODUCTION_REPAIR_ATTEMPTS_PER_RUN = 4 +const PRODUCTION_EPOCH_START_BUDGET_MS = 8 * 60 * 1000 const DEVELOPMENT_SYNCING_LEASE_DURATION_MS = 5 * 60 * 1000 export function getEpochCandidateLimit(isDevelopment: boolean): number { @@ -27,7 +29,7 @@ export function getSyncingLeaseDurationMs(isDevelopment: boolean): number { } export function isFullRepairAllowed(fullRepairAttempts: number, isDevelopment: boolean): boolean { - return isDevelopment || fullRepairAttempts === 0 + return isDevelopment || fullRepairAttempts < MAX_PRODUCTION_REPAIR_ATTEMPTS_PER_RUN } function formatError(error: unknown): string { @@ -59,6 +61,7 @@ export default defineTask({ }, async run() { const config = useSafeRuntimeConfig() + const startedAt = Date.now() try { const rpcUrl = getRpcUrl() @@ -97,11 +100,15 @@ export default defineTask({ let fullRepairAttempts = 0 for (const epochNumber of plannedEpochs) { + // Leave unstarted epochs eligible for the next run instead of claiming and failing them. + if (!isFullRepairAllowed(fullRepairAttempts, import.meta.dev) + || (!import.meta.dev && Date.now() - startedAt >= PRODUCTION_EPOCH_START_BUDGET_MS)) { + break + } + let outcome: EpochSyncOutcome try { - outcome = await synchronizeCompletedEpoch(epochNumber, config.public.nimiqNetwork, { - allowRepair: isFullRepairAllowed(fullRepairAttempts, import.meta.dev), - }) + outcome = await synchronizeCompletedEpoch(epochNumber, config.public.nimiqNetwork) } catch (error) { outcome = { epochNumber, status: 'failed', error: formatError(error), repairAttempted: false } @@ -114,6 +121,9 @@ export default defineTask({ fullRepairAttempts++ } + const deferredEpochs = plannedEpochs.slice(outcomes.length) + consola.info(`[sync:epochs] finalized ${epochsSynced.length}, repair attempts ${fullRepairAttempts}, deferred ${deferredEpochs.length} planned epoch(s)`) + const failures = outcomes.filter((outcome): outcome is Extract