Add publish rsr to smart contract

.github/workflows/ci.yml

@@ -102,7 +102,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: superfly/flyctl-actions/setup-flyctl@master
-      - run: flyctl deploy --remote-only
+      - run: flyctl deploy --remote-only —-build-arg GIT_COMMIT=$(git rev-parse HEAD)
         env:
           FLY_API_TOKEN: ${{ secrets.FLY_API_TOKEN }}
       - if: failure()

Dockerfile

@@ -14,6 +14,9 @@ ENV NODE_ENV=production
 # 4096MB available memory - 200MB for anything else
 ENV NODE_OPTIONS="--max-old-space-size=3896"
 
+ARG GIT_COMMIT
+ENV GIT_COMMIT=$GIT_COMMIT
+
 # Throw-away build stage to reduce size of final image
 FROM base AS build
 

bin/cancel-pending-tx.js

@@ -6,7 +6,7 @@ import { ethers } from 'ethers'
 import { CoinType, newDelegatedEthAddress } from '@glif/filecoin-address'
 import pRetry from 'p-retry'
 
-import { createMeridianContract } from '../lib/ie-contract.js'
+import { createContracts } from '../lib/contracts.js'
 
 const {
   WALLET_SEED
@@ -17,7 +17,7 @@ const [, , tx] = process.argv
 assert(WALLET_SEED, 'WALLET_SEED required')
 assert(tx, 'Transaction hash must be provided as the first argument')
 
-const { provider } = await createMeridianContract()
+const { provider } = createContracts()
 
 const signer = ethers.Wallet.fromPhrase(WALLET_SEED, provider)
 const walletDelegatedAddress = newDelegatedEthAddress(/** @type {any} */(signer.address), CoinType.MAIN).toString()

bin/dry-run.js

@@ -12,7 +12,7 @@ import path from 'node:path'
 import { fileURLToPath } from 'node:url'
 import pg from 'pg'
 import { RoundData } from '../lib/round.js'
-import { createMeridianContract } from '../lib/ie-contract.js'
+import { createContracts } from '../lib/contracts.js'
 import * as SparkImpactEvaluator from '@filecoin-station/spark-impact-evaluator'
 
 /** @typedef {import('../lib/preprocess.js').Measurement} Measurement */
@@ -142,7 +142,8 @@ const { ignoredErrors } = await evaluate({
   setScores,
   logger: console,
   recordTelemetry,
-  createPgClient
+  createPgClient,
+  gitCommit: ''
 })
 
 console.log('Duration: %sms', Date.now() - started)
@@ -213,7 +214,7 @@ async function fetchMeasurementsAddedEvents (contractAddress, roundIndex) {
  * @param {bigint} roundIndex
  */
 async function fetchMeasurementsAddedFromChain (contractAddress, roundIndex) {
-  const { ieContract, provider } = await createMeridianContract(contractAddress)
+  const { ieContract, provider } = createContracts(contractAddress)
 
   console.log('Fetching MeasurementsAdded events from the ledger')
 
@@ -268,6 +269,6 @@ function isEventLog (logOrEventLog) {
  * @param {string} contractAddress
  */
 async function fetchLastRoundIndex (contractAddress) {
-  const { ieContract } = await createMeridianContract(contractAddress)
+  const { ieContract } = await createContracts(contractAddress)
   return await ieContract.currentRoundIndex()
 }

bin/fetch-recent-miner-measurements.js

@@ -9,7 +9,7 @@ import { mkdir, readFile, writeFile } from 'node:fs/promises'
 import os from 'node:os'
 import path from 'node:path'
 import pMap from 'p-map'
-import { createMeridianContract } from '../lib/ie-contract.js'
+import { createContracts } from '../lib/contracts.js'
 import { fetchMeasurements, preprocess } from '../lib/preprocess.js'
 import { RoundData } from '../lib/round.js'
 import * as SparkImpactEvaluator from '@filecoin-station/spark-impact-evaluator'
@@ -140,7 +140,7 @@ console.error('Wrote human-readable summary for %s to %s', minerId, MINER_SUMMAR
  * @returns
  */
 async function getRecentMeasurementsAddedEvents (contractAddress, blocksToQuery = Number.POSITIVE_INFINITY) {
-  const { ieContract } = await createMeridianContract(contractAddress)
+  const { ieContract } = createContracts(contractAddress)
 
   // max look-back period allowed by Glif.io is 2000 blocks (approx 16h40m)
   // in practice, requests for the last 2000 blocks are usually rejected,
@@ -214,7 +214,8 @@ async function processRound (roundIndex, measurementCids, resultCounts) {
     recordTelemetry,
     logger: { log: debug, error: debug },
     ieContract,
-    setScores: async () => {}
+    setScores: async () => {},
+    gitCommit: ''
   })
 
   for (const m of round.measurements) {

bin/spark-evaluate.js

@@ -9,12 +9,20 @@ import { recordTelemetry } from '../lib/telemetry.js'
 import { fetchMeasurements } from '../lib/preprocess.js'
 import { migrateWithPgConfig } from '../lib/migrate.js'
 import pg from 'pg'
-import { createMeridianContract } from '../lib/ie-contract.js'
+import { createContracts } from '../lib/contracts.js'
 import { setScores } from '../lib/submit-scores.js'
+import { runPublishRsrLoop } from '../lib/publish-rsr.js'
+import * as Client from '@web3-storage/w3up-client'
+import { ed25519 } from '@ucanto/principal'
+import { CarReader } from '@ipld/car'
+import { importDAG } from '@ucanto/core/delegation'
 
 const {
   SENTRY_ENVIRONMENT = 'development',
-  WALLET_SEED
+  WALLET_SEED,
+  STORACHA_PRIVATE_KEY,
+  STORACHA_PROOF,
+  GIT_COMMIT
 } = process.env
 
 Sentry.init({
@@ -25,10 +33,27 @@ Sentry.init({
 })
 
 assert(WALLET_SEED, 'WALLET_SEED required')
+assert(STORACHA_PRIVATE_KEY, 'STORACHA_PRIVATE_KEY required')
+assert(STORACHA_PROOF, 'STORACHA_PROOF required')
 
 await migrateWithPgConfig({ connectionString: DATABASE_URL })
 
-const { ieContract, provider } = await createMeridianContract()
+async function parseProof (data) {
+  const blocks = []
+  const reader = await CarReader.fromBytes(Buffer.from(data, 'base64'))
+  for await (const block of reader.blocks()) {
+    blocks.push(block)
+  }
+  return importDAG(blocks)
+}
+
+const principal = ed25519.Signer.parse(STORACHA_PRIVATE_KEY)
+const storachaClient = await Client.create({ principal })
+const proof = await parseProof(STORACHA_PROOF)
+const space = await storachaClient.addSpace(proof)
+await storachaClient.setCurrentSpace(space.did())
+
+const { ieContract, rsrContract, provider } = createContracts()
 
 const signer = ethers.Wallet.fromPhrase(WALLET_SEED, provider)
 const walletDelegatedAddress = newDelegatedEthAddress(/** @type {any} */(signer.address), CoinType.MAIN).toString()
@@ -41,12 +66,20 @@ const createPgClient = async () => {
   return pgClient
 }
 
-await startEvaluate({
-  ieContract,
-  fetchMeasurements,
-  fetchRoundDetails,
-  recordTelemetry,
-  createPgClient,
-  logger: console,
-  setScores: (participants, values) => setScores(signer, participants, values)
-})
+await Promise.all([
+  startEvaluate({
+    ieContract,
+    fetchMeasurements,
+    fetchRoundDetails,
+    recordTelemetry,
+    createPgClient,
+    logger: console,
+    setScores: (participants, values) => setScores(signer, participants, values),
+    gitCommit: GIT_COMMIT
+  }),
+  runPublishRsrLoop({
+    createPgClient,
+    storachaClient,
+    rsrContract: rsrContract.connect(signer)
+  })
+])

index.js

@@ -19,7 +19,8 @@ export const startEvaluate = async ({
   recordTelemetry,
   createPgClient,
   logger,
-  setScores
+  setScores,
+  gitCommit
 }) => {
   assert(typeof createPgClient === 'function', 'createPgClient must be a function')
 
@@ -116,7 +117,8 @@ export const startEvaluate = async ({
       recordTelemetry,
       createPgClient,
       logger,
-      setScores
+      setScores,
+      gitCommit
     }).catch(err => {
       console.error('CANNOT EVALUATE ROUND %s:', evaluatedRoundIndex, err)
       Sentry.captureException(err, {

lib/ie-contract.js → lib/contracts.js

@@ -1,8 +1,17 @@
 import { ethers } from 'ethers'
 import { rpcUrls, GLIF_TOKEN } from './config.js'
 import * as SparkImpactEvaluator from '@filecoin-station/spark-impact-evaluator'
+import fs from 'node:fs/promises'
+import { fileURLToPath } from 'node:url'
 
-export const createMeridianContract = async (contractAddress = SparkImpactEvaluator.ADDRESS) => {
+const rsrContractAbi = JSON.parse(
+  await fs.readFile(
+    fileURLToPath(new URL('./rsrContract.json', import.meta.url)),
+    'utf8'
+  )
+).abi
+
+export const createContracts = (ieContractAddress = SparkImpactEvaluator.ADDRESS) => {
   const provider = new ethers.FallbackProvider(rpcUrls.map(url => {
     const fetchRequest = new ethers.FetchRequest(url)
     fetchRequest.setHeader('Authorization', `Bearer ${GLIF_TOKEN}`)
@@ -16,10 +25,16 @@ export const createMeridianContract = async (contractAddress = SparkImpactEvalua
   // provider.on('debug', d => console.log('[ethers:debug %s] %s %o', new Date().toISOString().split('T')[1], d.action, d.payload ?? d.result))
 
   const ieContract = new ethers.Contract(
-    contractAddress,
+    ieContractAddress,
     SparkImpactEvaluator.ABI,
     provider
   )
 
-  return { ieContract, provider }
+  const rsrContract = new ethers.Contract(
+    'TODO',
+    rsrContractAbi,
+    provider
+  )
+
+  return { ieContract, rsrContract, provider }
 }

lib/evaluate.js

@@ -7,6 +7,7 @@ import { buildRetrievalStats, getTaskId, recordCommitteeSizes } from './retrieva
 import { getTasksAllowedForStations } from './tasker.js'
 import { groupMeasurementsToCommittees } from './committee.js'
 import pRetry from 'p-retry'
+import * as publishRsr from './publish-rsr.js'
 
 /** @import {Measurement} from './preprocess.js' */
 
@@ -26,6 +27,7 @@ export const REQUIRED_COMMITTEE_SIZE = 30
  * @param {import('./typings.js').RecordTelemetryFn} args.recordTelemetry
  * @param {import('./typings.js').CreatePgClient} [args.createPgClient]
  * @param {Pick<Console, 'log' | 'error'>} args.logger
+ * @param {string} args.gitCommit
  */
 export const evaluate = async ({
   round,
@@ -36,7 +38,8 @@ export const evaluate = async ({
   fetchRoundDetails,
   recordTelemetry,
   createPgClient,
-  logger
+  logger,
+  gitCommit
 }) => {
   requiredCommitteeSize ??= REQUIRED_COMMITTEE_SIZE
 
@@ -47,6 +50,7 @@ export const evaluate = async ({
   // Detect fraud
 
   const sparkRoundDetails = await fetchRoundDetails(await ieContract.getAddress(), roundIndex, recordTelemetry)
+  round.details = sparkRoundDetails
   // Omit the roundDetails object from the format string to get nicer formatting
   debug('ROUND DETAILS for round=%s', roundIndex, sparkRoundDetails)
 
@@ -202,6 +206,14 @@ export const evaluate = async ({
     }
   }
 
+  try {
+    await publishRsr.postEvaluate({ createPgClient, round, gitCommit })
+  } catch (err) {
+    console.error('Cannot store data for `spark-rsr-contract`.', err)
+    ignoredErrors.push(err)
+    Sentry.captureException(err)
+  }
+
   return { ignoredErrors }
 }
 

lib/preprocess.js

@@ -116,6 +116,7 @@ export const preprocess = async ({
   logger.log('Retrieval Success Rate: %s%s (%s of %s)', Math.round(100 * okCount / total), '%', okCount, total)
 
   round.measurements.push(...validMeasurements)
+  round.cids.push(cid)
 
   recordTelemetry('preprocess', point => {
     point.intField('round_index', roundIndex)

lib/publish-rsr.js

@@ -0,0 +1,96 @@
+import * as Sentry from '@sentry/node'
+import timers from 'node:timers/promises'
+import pRetry from 'p-retry'
+
+const ONE_HOUR = 60 * 60 * 1000
+
+const withPgClient = fn => async ({ createPgClient, ...args }) => {
+  const pgClient = await createPgClient()
+  try {
+    return await fn({ pgClient, ...args })
+  } finally {
+    await pgClient.end()
+  }
+}
+
+export const postEvaluate = withPgClient(async ({ pgClient, round, gitCommit }) => {
+  await pgClient.query(`
+    INSERT INTO publish_rsr_rounds
+    (round_index, spark_evaluate_version, measurement_commitments, round_details, providers)
+    VALUES
+    ($1, $2, $3, $4, $5)
+  `, [
+    round.index,
+    gitCommit,
+    round.cids,
+    round.details,
+    round.measurements.reduce((acc, m) => {
+      acc[m.minerId] = acc[m.minerId] || { successful: 0, total: 0 }
+      acc[m.minerId].total++
+      if (m.retrievalResult === 'OK') {
+        acc[m.minerId].successful++
+      }
+      return acc
+    }, {})
+  ])
+})
+
+const publishRsr = withPgClient(async ({ pgClient, storachaClient, rsrContract }) => {
+  const { rows } = await pgClient.query(`
+    SELECT *
+    FROM publish_rsr_rounds
+    WHERE evaluated_at <= now()::date AND evaluated_at > now()::date - interval '1 day'
+    ORDER BY round_index
+  `)
+
+  const providers = new Map()
+  for (const row of rows) {
+    for (const [minerId, { successful, total }] of Object.entries(row.providers)) {
+      if (!providers.has(minerId)) {
+        providers.set(minerId, { successful: 0, total: 0 })
+      }
+      providers.get(minerId).successful += successful
+      providers.get(minerId).total += total
+    }
+  }
+
+  const directoryCid = await pRetry(() => storachaClient.uploadDirectory([
+    new File([JSON.stringify({
+      date: rows[0].evaluated_at.toISOString().split('T')[0],
+      meta: {
+        rounds: Object.fromEntries(rows.map(row => [
+          row.round_index,
+          {
+            sparkEvaluateVersion: row.spark_evaluate_version,
+            measurementCommitments: row.measurement_commitments,
+            roundDetails: row.round_details
+          }
+        ]))
+      },
+      providers
+    })], 'commitment.json')
+  ]))
+  console.log(`https://${directoryCid}.ipfs.w3s.link/commitment.json`)
+
+  const tx = await pRetry(() => rsrContract.addCommitment(directoryCid.toString()))
+  console.log(tx.hash)
+  await tx.wait()
+
+  await pgClient.query(`
+    DELETE FROM publish_rsr_rounds
+    WHERE round_index < $1
+  `, [rows[0].round_index])
+})
+
+export const runPublishRsrLoop = async ({ createPgClient, storachaClient, rsrContract }) => {
+  while (true) {
+    try {
+      await publishRsr({ createPgClient, storachaClient, rsrContract })
+    } catch (err) {
+      console.error(err)
+      Sentry.captureException(err)
+    } finally {
+      await timers.setTimeout(ONE_HOUR)
+    }
+  }
+}