fix: data sync + enable observability

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>> => {

scripts/validate-json-files.ts

@@ -1,6 +1,5 @@
 import type { Result } from 'nimiq-validator-trustscore/types'
 import type { ValidatorJSON } from '../server/utils/schemas'
-import { execSync } from 'node:child_process'
 import process from 'node:process'
 import { consola } from 'consola'
 import { importValidators } from '../server/utils/json-files'
@@ -32,10 +31,9 @@ function formatValidationErrors(error: any): ValidationErrorDetails[] {
   })
 }
 
-function logValidationErrors(errors: ValidationErrorDetails[], source: string, network: string) {
-  consola.error(`\n❌ Validation failed for ${network} validators from ${source}:\n`)
+function logValidationErrors(errors: ValidationErrorDetails[], network: string) {
+  consola.error(`\n❌ Validation failed for ${network} validators:\n`)
 
-  // Group errors by validator index
   const errorsByValidator = errors.reduce((acc, error) => {
     if (!acc[error.validatorIndex]) {
       acc[error.validatorIndex] = []
@@ -56,89 +54,47 @@ function logValidationErrors(errors: ValidationErrorDetails[], source: string, n
         consola.error(`      Current value: ${valueStr}`)
       }
     })
-    consola.error('') // Empty line between validators
+    consola.error('')
   })
 
   consola.info(`\n💡 Tips for fixing these errors:`)
 
   const uniqueFields = [...new Set(errors.map(e => e.field))]
-  if (uniqueFields.includes('address')) {
+  if (uniqueFields.includes('address'))
     consola.info(`   • Nimiq addresses should follow format: "NQ## #### #### #### #### #### #### #### ####"`)
-  }
-  if (uniqueFields.includes('logo')) {
+  if (uniqueFields.includes('logo'))
     consola.info(`   • Logos should be data URLs starting with "data:image/png,", "data:image/svg+xml," or "data:image/webp,"`)
-  }
-  if (uniqueFields.some(f => f.includes('contact'))) {
+  if (uniqueFields.some(f => f.includes('contact')))
     consola.info(`   • Social media handles should not include '@' symbol or should follow platform-specific format rules`)
-  }
-  if (uniqueFields.includes('website')) {
+  if (uniqueFields.includes('website'))
     consola.info(`   • Websites must be valid URLs starting with http:// or https://`)
-  }
 }
 
-async function validateValidators(source: 'filesystem' | 'github', nimiqNetwork: string, gitBranch: string): Result<ValidatorJSON[]> {
-  const [importOk, errorReading, _validators] = await importValidators(source, { nimiqNetwork, shouldStore: false, gitBranch })
+async function validateValidators(nimiqNetwork: string): Result<ValidatorJSON[]> {
+  const [importOk, errorReading, _validators] = await importValidators(nimiqNetwork)
   if (!importOk)
     return [false, errorReading, undefined]
 
   const result = validatorsSchema.safeParse(_validators)
   if (!result.success) {
     const errors = formatValidationErrors(result.error)
-    logValidationErrors(errors, source, nimiqNetwork)
+    logValidationErrors(errors, nimiqNetwork)
     return [false, `Found ${errors.length} validation error(s)`, undefined]
   }
 
   return [true, undefined, result.data]
 }
 
-// get flag from --source. should be either 'filesystem' or 'github'
-const args = process.argv.slice(2)
-const source = args[0] || 'filesystem'
-if (source !== 'filesystem' && source !== 'github') {
-  consola.error('Invalid source. Use either "filesystem" or "github".')
-  process.exit(1)
-}
-consola.info(`Validating validators from ${source}...`)
-
-// Try to get git branch, with fallbacks for GitHub Actions environment
-let gitBranch: string = ''
-
-try {
-  gitBranch = execSync('git branch --show-current', { stdio: ['ignore', 'pipe', 'ignore'] }).toString().trim()
-}
-catch {
-  // Fallback for GitHub Actions - git command might fail in detached HEAD state
-  consola.warn('Unable to get current branch from git command, trying environment variables...')
-}
-
-// If git command failed or returned empty, try GitHub Actions environment variables
-if (!gitBranch) {
-  // For pull requests, use the head ref (source branch)
-  gitBranch = process.env.GITHUB_HEAD_REF
-  // For pushes, use the ref name
-    || process.env.GITHUB_REF_NAME
-  // Last resort: use the commit SHA
-    || process.env.GITHUB_SHA
-    || ''
-}
-
-if (!gitBranch) {
-  consola.error('Unable to determine git branch or commit reference. Make sure you are in a git repository or running in a GitHub Actions environment.')
-  process.exit(1)
-}
-
-consola.info(`Using git reference: ${gitBranch}`)
-
-const [okMain, errorMain, validatorsMain] = await validateValidators(source, 'main-albatross', gitBranch)
+const [okMain, errorMain, validatorsMain] = await validateValidators('main-albatross')
 if (!okMain) {
   consola.error(errorMain)
   process.exit(1)
 }
-consola.success(`✅ All ${validatorsMain.length} validators for main-albatross are valid in ${source}!`)
+consola.success(`✅ All ${validatorsMain.length} validators for main-albatross are valid!`)
 
-const [okTest, errorTest, validatorsTest] = await validateValidators(source, 'test-albatross', gitBranch)
+const [okTest, errorTest, validatorsTest] = await validateValidators('test-albatross')
 if (!okTest) {
   consola.error(errorTest)
   process.exit(1)
 }
-consola.success(`✅ All ${validatorsTest.length} validators for test-albatross are valid in ${source}!`)
+consola.success(`✅ All ${validatorsTest.length} validators for test-albatross are valid!`)

server/assets/validators

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

server/plugins/setup-database.ts

@@ -10,7 +10,7 @@ export default defineNitroPlugin(async () => {
     if (gitBranch !== 'dev')
       return
 
-    const [ok, error, validators] = await importValidators('filesystem', { nimiqNetwork, gitBranch })
+    const [ok, error, validators] = await importValidatorsBundled(nimiqNetwork)
     if (!ok)
       throw new Error(`Error importing validators: ${error}`)
     consola.success(`${validators.length} validators imported successfully`)

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/tasks/sync/snapshot.ts

@@ -16,13 +16,11 @@ export default defineTask({
         throw new Error('No Albatross RPC Node URL')
       initRpcClient({ url: rpcUrl })
 
-      const { nimiqNetwork, gitBranch } = config.public
+      const { nimiqNetwork } = config.public
 
-      const [importSuccess, errorImport, importData] = import.meta.dev
-        ? await importValidators('filesystem', { nimiqNetwork, gitBranch })
-        : await importValidatorsBundled(nimiqNetwork)
+      const [importSuccess, errorImport, importData] = await importValidatorsBundled(nimiqNetwork)
       if (!importSuccess || !importData) {
-        const error = new Error(errorImport || 'Unable to import from GitHub')
+        const error = new Error(errorImport || 'Unable to import validators')
         await sendSyncFailureNotification('snapshot', error)
         return { result: { success: false, error: errorImport } }
       }

server/utils/json-files.ts

@@ -2,133 +2,41 @@ import type { Result } from 'nimiq-validator-trustscore/types'
 import type { ValidatorJSON } from './schemas'
 import { readdir, readFile } from 'node:fs/promises'
 import { extname } from 'node:path'
-import process from 'node:process'
-import { consola } from 'consola'
-import { $fetch } from 'ofetch'
 import { join } from 'pathe'
 import { validatorSchema } from './schemas'
+
 /**
  * Import validators from a folder containing .json files.
- *
- * This function is expected to be used when initializing the database with validators, so it will throw
- * an error if the files are not valid and the program should stop.
+ * Used by the validation script to check files on disk.
  */
-export async function importValidatorsFromFiles(folderPath: string): Result<any[]> {
+export async function importValidators(nimiqNetwork: string): Result<ValidatorJSON[]> {
+  if (!nimiqNetwork)
+    return [false, 'Nimiq network is required', undefined]
+
+  const folderPath = `public/validators/${nimiqNetwork}`
   const allFiles = await readdir(folderPath)
   const files = allFiles
     .filter(f => extname(f) === '.json')
     .filter(f => !f.endsWith('.example.json'))
 
-  const rawValidators: any[] = []
+  const validators: ValidatorJSON[] = []
   for (const file of files) {
     const filePath = join(folderPath, file)
     const fileContent = await readFile(filePath, 'utf8')
 
+    let raw: unknown
     try {
-      rawValidators.push(JSON.parse(fileContent))
+      raw = JSON.parse(fileContent)
     }
     catch (error) {
       return [false, `Invalid JSON in file: ${file}. Error: ${error}`, undefined]
     }
-  }
-  return [true, undefined, rawValidators]
-}
-
-/**
- * Import validators from GitHub using the official GitHub REST API.
- */
-async function importValidatorsFromGitHub(path: string, { gitBranch }: Pick<ImportValidatorsFromFilesOptions, 'gitBranch'>): Result<any[]> {
-  // Default to main branch if not specified
-  const branch = gitBranch || 'main'
-
-  // Check if running in a fork PR (GitHub Actions sets GITHUB_HEAD_REPOSITORY)
-  const headRepo = process.env.GITHUB_HEAD_REPOSITORY // format: "owner/repo"
-  const baseRepo = process.env.GITHUB_REPOSITORY || 'nimiq/validators-api'
-
-  // Use head repo if it's different from base (fork PR scenario)
-  const [owner, repo] = (headRepo && headRepo !== baseRepo ? headRepo : baseRepo).split('/')
-
-  const apiUrl = `https://api.github.com/repos/${owner}/${repo}/contents/${path}?ref=${branch}`
-
-  // 1. List directory contents
-  let listing: Array<{
-    name: string
-    path: string
-    type: 'file' | 'dir'
-    download_url: string
-  }>
-
-  const headers: HeadersInit = { 'User-Agent': 'request' }
-  try {
-    listing = await $fetch(apiUrl, { headers })
-  }
-  catch (e) {
-    consola.warn(`Error listing validators folder on GitHub: ${e}`)
-    return [false, `Could not list validators on GitHub ${apiUrl} | ${e}`, undefined]
-  }
-
-  // 2. Filter only .json files (skip .example.json)
-  const jsonFiles = listing.filter(file =>
-    file.type === 'file'
-    && file.name.endsWith('.json')
-    && !file.name.endsWith('.example.json'),
-  )
-
-  // 3. Fetch each file’s raw contents
-  const rawContents = await Promise.all(jsonFiles.map(async (file) => {
-    try {
-      return await $fetch<string>(file.download_url, { headers })
-    }
-    catch (e) {
-      consola.warn(`Failed to download ${file.path}: ${e}`)
-      return [false, `Failed to download ${file.path}: ${e}`, undefined]
-    }
-  }))
-
-  // 4. Parse JSON and return
-  const parsed = rawContents.filter((c): c is string => Boolean(c)).map(c => JSON.parse(c!))
-
-  return [true, undefined, parsed]
-}
-
-interface ImportValidatorsFromFilesOptions {
-  nimiqNetwork?: string
-  gitBranch?: string
-  shouldStore?: boolean
-}
 
-/**
- * Import validators from either the filesystem or GitHub, then validate & store.
- */
-export async function importValidators(source: 'filesystem' | 'github', options: ImportValidatorsFromFilesOptions = {}): Result<ValidatorJSON[]> {
-  const { nimiqNetwork, shouldStore = true, gitBranch } = options
-  if (!nimiqNetwork)
-    return [false, 'Nimiq network is required', undefined]
-
-  const path = `public/validators/${nimiqNetwork}`
-
-  const [ok, readError, data] = source === 'filesystem'
-    ? await importValidatorsFromFiles(path)
-    : await importValidatorsFromGitHub(path, { gitBranch })
-
-  if (!ok)
-    return [false, readError, undefined]
-
-  const validators: ValidatorJSON[] = []
-  for (const validator of data) {
-    const { data, success, error } = validatorSchema.safeParse(validator)
-    if (!success)
-      return [false, `Invalid validator ${validator.name}(${validator.address}) data: ${error || 'Unknown error'}. ${JSON.stringify({ path, gitBranch, source })}`, undefined]
-    validators.push(data)
+    const parsed = validatorSchema.safeParse(raw)
+    if (!parsed.success)
+      return [false, `Invalid validator ${file}: ${parsed.error}`, undefined]
+    validators.push(parsed.data)
   }
 
-  if (!shouldStore)
-    return [true, undefined, validators]
-
-  const results = await Promise.allSettled(validators.map(v => storeValidator(v.address, v, { upsert: true })))
-  const failures = results.filter(r => r.status === 'rejected')
-  if (failures.length > 0)
-    return [false, `Errors importing validators: ${failures.map((f: any) => f.reason).join(', ')}`, undefined]
-
   return [true, undefined, validators]
 }