fix: data sync + enable observability

Author: onmax

packages/nimiq-validator-trustscore/src/fetcher.ts

@@ -11,14 +11,19 @@ export interface FetchActivityOptions extends Pick<BaseAlbatrossPolicyOptions, '
    * @default 5
    */
   maxRetries?: number
+  /**
+   * Max concurrent RPC requests per batch. Workers allows ~6 concurrent outbound connections.
+   * @default 120
+   */
+  maxBatchSize?: number
 }
 
 /**
  * For a given block number, fetches the validator slots assignation.
  * The block number MUST be an election block otherwise it will return an error result.
  */
 export async function fetchActivity(epochIndex: number, options: FetchActivityOptions = {}): Result<EpochActivity> {
-  const { maxRetries = 5, network = 'mainnet' } = options
+  const { maxRetries = 5, network = 'mainnet', maxBatchSize: _maxBatchSize = 120 } = options
   // Epochs start at 1, but election block is the first block of the epoch
   const electionBlock = electionBlockOf(epochIndex, { network })!
   const [isBlockOk, errorBlockNumber, block] = await getBlockByNumber({ blockNumber: electionBlock, includeBody: false })
@@ -51,8 +56,8 @@ export async function fetchActivity(epochIndex: number, options: FetchActivityOp
     epochActivity[validator] = { address: validator, likelihood, missed: 0, rewarded: 0, dominanceRatioViaBalance, dominanceRatioViaSlots, balance, elected: true, stakers: 0 } as ElectedValidator
   }
 
-  const maxBatchSize = 120
-  const minBatchSize = 10
+  const maxBatchSize = _maxBatchSize
+  const minBatchSize = Math.min(10, maxBatchSize)
   let batchSize = maxBatchSize
 
   const createPromise = async (index: number, retryCount = 0): Promise<ResultSync<void>> => {

server/assets/validators

@@ -0,0 +1 @@
+../../public/validators

server/tasks/cron/sync.ts

@@ -52,7 +52,10 @@ export default defineTask({
       for (const taskName of TASKS) {
         consola.info(`[cron:sync] running ${taskName}`)
         const res = await runTask(taskName, { payload: event.payload ?? {}, context: event.context ?? {} })
-        results[taskName] = (res as any)?.result ?? res
+        const result = (res as any)?.result ?? res
+        results[taskName] = result
+        if (result?.success === false)
+          throw new Error(`${taskName} failed: ${result.error || 'unknown'}`)
       }
 
       if (cronRunId) {

server/tasks/sync/epochs.ts

@@ -39,7 +39,7 @@ export default defineTask({
           break
 
         const missingEpoch = missingEpochs.at(0)!
-        const [activityOk, activityError, epochActivity] = await fetchActivity(missingEpoch)
+        const [activityOk, activityError, epochActivity] = await fetchActivity(missingEpoch, { network: config.public.nimiqNetwork, maxBatchSize: import.meta.dev ? 120 : 6 })
         if (!activityOk || !epochActivity) {
           const error = new Error(activityError || 'Unable to fetch activity')
           await sendSyncFailureNotification('missing-epoch', error)

server/utils/validators-bundle.ts

@@ -7,46 +7,31 @@ interface ImportValidatorsBundledOptions {
   shouldStore?: boolean
 }
 
-function normalizePath(value: string) {
-  return value.replace(/\\/g, '/')
-}
-
-function extractNetworkFromPath(path: string) {
-  const normalized = normalizePath(path)
-  const match = normalized.match(/(?:^|\/)public\/validators\/([^/]+)\/[^/]+\.json$/)
-  return match?.[1]
-}
-
 export async function importValidatorsBundled(nimiqNetwork?: string, options: ImportValidatorsBundledOptions = {}): Result<ValidatorJSON[]> {
   if (!nimiqNetwork)
     return [false, 'Nimiq network is required', undefined]
 
   const { shouldStore = true } = options
-  const modules = import.meta.glob('../../public/validators/*/*.json', { eager: true, import: 'default' })
+  const storage = useStorage('assets:server:validators')
+  const keys = await storage.getKeys(`${nimiqNetwork}`)
 
   const validators: ValidatorJSON[] = []
-  for (const [path, mod] of Object.entries(modules)) {
-    if (path.endsWith('.example.json'))
+  for (const key of keys) {
+    if (!key.endsWith('.json') || key.endsWith('.example.json'))
       continue
 
-    const network = extractNetworkFromPath(path)
-    if (!network || network !== nimiqNetwork)
-      continue
-
-    const data = (mod as any)?.default ?? mod
+    const data = await storage.getItem(key)
     const parsed = validatorSchema.safeParse(data)
-    if (!parsed.success) {
-      return [false, `Invalid validator data at ${path}: ${parsed.error}`, undefined]
-    }
+    if (!parsed.success)
+      return [false, `Invalid validator data at ${key}: ${parsed.error}`, undefined]
     validators.push(parsed.data)
   }
 
   if (!shouldStore)
     return [true, undefined, validators]
 
-  if (validators.length === 0) {
+  if (validators.length === 0)
     return [false, `No bundled validators found for network: ${nimiqNetwork}`, undefined]
-  }
 
   const results = await Promise.allSettled(validators.map(v => storeValidator(v.address, v, { upsert: true })))
   const failures = results.filter(r => r.status === 'rejected')

wrangler.json

@@ -9,7 +9,7 @@
   "observability": {
     "enabled": true,
     "logs": {
-      "enabled": false,
+      "enabled": true,
       "invocation_logs": true
     }
   },
@@ -24,6 +24,7 @@
       "main": "dist/server/index.mjs",
       "assets": { "directory": "dist/public" },
       "compatibility_flags": ["nodejs_compat"],
+      "observability": { "enabled": true, "logs": { "enabled": true, "invocation_logs": true } },
       "triggers": { "crons": ["0 */12 * * *"] },
       "vars": { "NUXT_PUBLIC_NIMIQ_NETWORK": "main-albatross" },
       "d1_databases": [{ "binding": "DB", "database_id": "dd9fa8b7-f5a1-48ab-b0af-d93d1092b89a", "database_name": "validators-api-mainnet-preview" }],
@@ -35,6 +36,7 @@
       "main": "dist/server/index.mjs",
       "assets": { "directory": "dist/public" },
       "compatibility_flags": ["nodejs_compat"],
+      "observability": { "enabled": true, "logs": { "enabled": true, "invocation_logs": true } },
       "triggers": { "crons": ["0 */12 * * *"] },
       "vars": { "NUXT_PUBLIC_NIMIQ_NETWORK": "test-albatross" },
       "d1_databases": [{ "binding": "DB", "database_id": "de14e353-5028-4e52-a383-a9cc200d960d", "database_name": "validators-api-testnet" }],
@@ -46,6 +48,7 @@
       "main": "dist/server/index.mjs",
       "assets": { "directory": "dist/public" },
       "compatibility_flags": ["nodejs_compat"],
+      "observability": { "enabled": true, "logs": { "enabled": true, "invocation_logs": true } },
       "triggers": { "crons": ["0 */12 * * *"] },
       "vars": { "NUXT_PUBLIC_NIMIQ_NETWORK": "test-albatross" },
       "d1_databases": [{ "binding": "DB", "database_id": "da6931b2-6d06-4598-911b-fba1fa19efd2", "database_name": "validators-api-testnet-preview" }],