refactor: break fraudAssessment into evaluations

bin/dry-run.js

@@ -148,7 +148,7 @@ if (ignoredErrors.length) {
 }
 
 if (DUMP) {
-  const props = ['cid', 'minerId', 'participantAddress', 'inet_group', 'retrievalResult', 'fraudAssessment']
+  const props = ['cid', 'minerId', 'participantAddress', 'inet_group', 'retrievalResult', 'taskingEvaluation', 'majorityEvaluation']
   for (const k of Object.keys(round.measurements[0])) {
     if (!props.includes(k)) props.push(k)
   }

bin/evaluate-measurements.js

@@ -42,7 +42,7 @@ const EVALUATION_NDJSON_FILE = `${basename(measurementsPath, '.ndjson')}.evaluat
 const evaluationTxtWriter = fs.createWriteStream(EVALUATION_TXT_FILE)
 const evaluationNdjsonWriter = fs.createWriteStream(EVALUATION_NDJSON_FILE)
 
-evaluationTxtWriter.write(formatHeader({ includeFraudAssesment: keepRejected }) + '\n')
+evaluationTxtWriter.write(formatHeader({ includeEvaluation: keepRejected }) + '\n')
 
 const resultCounts = {
   total: 0
@@ -98,22 +98,26 @@ async function processRound (roundIndex, measurements, resultCounts) {
   })
 
   for (const m of round.measurements) {
-    if (m.fraudAssessment !== 'OK') continue
+    // FIXME: we should include non-majority measurements too
+    // See https://github.com/filecoin-station/spark-evaluate/pull/396
+    if (m.taskingEvaluation !== 'OK' && m.majorityEvaluation === 'OK') continue
     resultCounts.total++
     resultCounts[m.retrievalResult] = (resultCounts[m.retrievalResult] ?? 0) + 1
   }
 
   if (!keepRejected) {
     round.measurements = round.measurements
       // Keep accepted measurements only
-      .filter(m => m.fraudAssessment === 'OK')
-      // Remove the fraudAssessment field as all accepted measurements have the same 'OK' value
-      .map(m => ({ ...m, fraudAssessment: undefined }))
+      // FIXME: we should include non-majority measurements too
+      // See https://github.com/filecoin-station/spark-evaluate/pull/396
+      .filter(m => m.taskingEvaluation === 'OK' && m.majorityEvaluation === 'OK')
+      // Remove the taskingEvaluation and majorityEvaluation fields as all accepted measurements have the same 'OK' value
+      .map(m => ({ ...m, taskingEvaluation: undefined, majorityEvaluation: undefined }))
   }
 
   evaluationTxtWriter.write(
     round.measurements
-      .map(m => formatMeasurement(m, { includeFraudAssesment: keepRejected }) + '\n')
+      .map(m => formatMeasurement(m, { includeEvaluation: keepRejected }) + '\n')
       .join('')
   )
   evaluationNdjsonWriter.write(
@@ -144,17 +148,19 @@ function isFlagEnabled (envVarValue) {
 /**
  * @param {import('../lib/preprocess.js').Measurement} m
  * @param {object} options
- * @param {boolean} [options.includeFraudAssesment]
+ * @param {boolean} [options.includeEvaluation]
  */
-function formatMeasurement (m, { includeFraudAssesment } = {}) {
+function formatMeasurement (m, { includeEvaluation } = {}) {
   const fields = [
     new Date(m.finished_at).toISOString(),
     (m.cid ?? '').padEnd(70),
     (m.protocol ?? '').padEnd(10)
   ]
 
-  if (includeFraudAssesment) {
-    fields.push((m.fraudAssessment === 'OK' ? '🫡  ' : '🙅  '))
+  if (includeEvaluation) {
+    // FIXME: we should distinguish tasking and majority evaluation
+    // See https://github.com/filecoin-station/spark-evaluate/pull/396
+    fields.push((m.taskingEvaluation === 'OK' && m.majorityEvaluation === 'OK' ? '🫡  ' : '🙅  '))
   }
 
   fields.push((m.retrievalResult ?? ''))
@@ -164,16 +170,16 @@ function formatMeasurement (m, { includeFraudAssesment } = {}) {
 
 /**
  * @param {object} options
- * @param {boolean} [options.includeFraudAssesment]
+ * @param {boolean} [options.includeEvaluation]
  */
-function formatHeader ({ includeFraudAssesment } = {}) {
+function formatHeader ({ includeEvaluation } = {}) {
   const fields = [
     'Timestamp'.padEnd(new Date().toISOString().length),
     'CID'.padEnd(70),
     'Protocol'.padEnd(10)
   ]
 
-  if (includeFraudAssesment) {
+  if (includeEvaluation) {
     fields.push('🕵️  ')
   }
 

lib/committee.js

@@ -48,7 +48,7 @@ export class Committee {
   addMeasurement (m) {
     assert.strictEqual(m.cid, this.retrievalTask.cid, 'cid must match')
     assert.strictEqual(m.minerId, this.retrievalTask.minerId, 'minerId must match')
-    assert.strictEqual(m.fraudAssessment, 'OK', 'only accepted measurements can be added')
+    assert.strictEqual(m.taskingEvaluation, 'OK', 'only measurements accepted by task evaluation can be added')
     this.#measurements.push(m)
   }
 
@@ -75,7 +75,7 @@ export class Committee {
         hasRetrievalMajority: false,
         retrievalResult: 'COMMITTEE_TOO_SMALL'
       }
-      for (const m of this.#measurements) m.fraudAssessment = 'COMMITTEE_TOO_SMALL'
+      for (const m of this.#measurements) m.majorityEvaluation = 'COMMITTEE_TOO_SMALL'
       return
     }
 
@@ -115,12 +115,15 @@ export class Committee {
     let retrievalResult
     if (retrievalResultMajority) {
       retrievalResult = retrievalResultMajority.majorityValue.retrievalResult
+      for (const m of retrievalResultMajority.majorityMeasurements) {
+        m.majorityEvaluation = 'OK'
+      }
       for (const m of retrievalResultMajority.minorityMeasurements) {
-        m.fraudAssessment = 'MINORITY_RESULT'
+        m.majorityEvaluation = 'MINORITY_RESULT'
       }
     } else {
       retrievalResult = 'MAJORITY_NOT_FOUND'
-      for (const m of this.#measurements) m.fraudAssessment = 'MAJORITY_NOT_FOUND'
+      for (const m of this.#measurements) m.majorityEvaluation = 'MAJORITY_NOT_FOUND'
     }
 
     this.evaluation = {

lib/evaluate.js

@@ -62,12 +62,14 @@ export const evaluate = async ({
     requiredCommitteeSize,
     logger
   })
-  const honestMeasurements = measurements.filter(m => m.fraudAssessment === 'OK')
+  const measurementsToReward = measurements.filter(
+    m => m.taskingEvaluation === 'OK' && m.majorityEvaluation === 'OK'
+  )
 
   // Calculate reward shares
   const participants = {}
   let sum = 0n
-  for (const measurement of honestMeasurements) {
+  for (const measurement of measurementsToReward) {
     if (!participants[measurement.participantAddress]) {
       participants[measurement.participantAddress] = 0n
     }
@@ -76,7 +78,7 @@ export const evaluate = async ({
   for (const [participantAddress, participantTotal] of Object.entries(participants)) {
     const score = participantTotal *
       MAX_SCORE /
-      BigInt(honestMeasurements.length)
+      BigInt(measurementsToReward.length)
     participants[participantAddress] = score
     sum += score
   }
@@ -88,24 +90,24 @@ export const evaluate = async ({
     logger.log('EVALUATE ROUND %s: added %s as rounding to MAX_SCORE', roundIndex, delta)
   }
 
-  // Calculate aggregates per fraud detection outcome
+  // Calculate aggregates per evaluation outcome
   // This is used for logging and telemetry
-  /** @type {Partial<Record<import('./typings.js').FraudAssesment, number>>} */
-  const fraudAssessments = {
+  /** @type {Partial<Record<import('./typings.js').TaskingEvaluation | import('./typings.js').CommitteeCheckError, number>>} */
+  const evaluationOutcomes = {
     OK: 0,
     TASK_NOT_IN_ROUND: 0,
     DUP_INET_GROUP: 0,
     TOO_MANY_TASKS: 0
   }
   for (const m of measurements) {
-    fraudAssessments[m.fraudAssessment] = (fraudAssessments[m.fraudAssessment] ?? 0) + 1
+    evaluationOutcomes[m.taskingEvaluation] = (evaluationOutcomes[m.taskingEvaluation] ?? 0) + 1
   }
   logger.log(
-    'EVALUATE ROUND %s: Evaluated %s measurements, found %s honest entries.\n%o',
+    'EVALUATE ROUND %s: Evaluated %s measurements, rewarding %s entries.\n%o',
     roundIndex,
     measurements.length,
-    honestMeasurements.length,
-    fraudAssessments
+    measurementsToReward.length,
+    evaluationOutcomes
   )
 
   const fraudDetectionDuration = Date.now() - started
@@ -138,11 +140,11 @@ export const evaluate = async ({
     point.intField('total_participants', Object.keys(participants).length)
     point.intField('total_measurements', measurements.length)
     point.intField('total_nodes', countUniqueNodes(measurements))
-    point.intField('honest_measurements', honestMeasurements.length)
+    point.intField('honest_measurements', measurementsToReward.length)
     point.intField('set_scores_duration_ms', setScoresDuration)
     point.intField('fraud_detection_duration_ms', fraudDetectionDuration)
 
-    for (const [type, count] of Object.entries(fraudAssessments)) {
+    for (const [type, count] of Object.entries(evaluationOutcomes)) {
       point.intField(`measurements_${type}`, count)
     }
   })
@@ -152,7 +154,9 @@ export const evaluate = async ({
   try {
     recordTelemetry('retrieval_stats_honest', (point) => {
       point.intField('round_index', roundIndex)
-      buildRetrievalStats(honestMeasurements, point)
+      // FIXME: Include non-majority measurements in these stats
+      // See https://github.com/filecoin-station/spark-evaluate/issues/446
+      buildRetrievalStats(measurementsToReward, point)
     })
   } catch (err) {
     console.error('Cannot record retrieval stats (honest).', err)
@@ -271,8 +275,6 @@ export const runFraudDetection = async ({
   //    or missing some of the required fields like `inet_group`
   //
   for (const m of measurements) {
-    if (m.fraudAssessment) continue
-
     // sanity checks to get nicer errors if we forget to set required fields in unit tests
     assert(typeof m.inet_group === 'string', 'missing inet_group')
     assert(typeof m.finished_at === 'number', 'missing finished_at')
@@ -281,15 +283,15 @@ export const runFraudDetection = async ({
       t => t.cid === m.cid && t.minerId === m.minerId
     )
     if (!isValidTask) {
-      m.fraudAssessment = 'TASK_NOT_IN_ROUND'
+      m.taskingEvaluation = 'TASK_NOT_IN_ROUND'
       continue
     }
 
     const isValidTaskForNode = tasksAllowedForStations.get(m.stationId).some(
       t => t.cid === m.cid && t.minerId === m.minerId
     )
     if (!isValidTaskForNode) {
-      m.fraudAssessment = 'TASK_WRONG_NODE'
+      m.taskingEvaluation = 'TASK_WRONG_NODE'
     }
   }
 
@@ -299,7 +301,7 @@ export const runFraudDetection = async ({
   /** @type {Map<string, Measurement[]>} */
   const inetGroups = new Map()
   for (const m of measurements) {
-    if (m.fraudAssessment) continue
+    if (m.taskingEvaluation) continue
 
     const key = m.inet_group
     let group = inetGroups.get(key)
@@ -347,18 +349,18 @@ export const runFraudDetection = async ({
 
       if (tasksSeen.has(taskId)) {
         debug('  pa: %s h: %s task: %s - task was already rewarded', m.participantAddress, h, taskId)
-        m.fraudAssessment = 'DUP_INET_GROUP'
+        m.taskingEvaluation = 'DUP_INET_GROUP'
         continue
       }
 
       if (tasksSeen.size >= sparkRoundDetails.maxTasksPerNode) {
         debug('  pa: %s h: %s task: %s - already rewarded max tasks', m.participantAddress, h, taskId)
-        m.fraudAssessment = 'TOO_MANY_TASKS'
+        m.taskingEvaluation = 'TOO_MANY_TASKS'
         continue
       }
 
       tasksSeen.add(taskId)
-      m.fraudAssessment = 'OK'
+      m.taskingEvaluation = 'OK'
       debug('  pa: %s h: %s task: %s - REWARD', m.participantAddress, h, taskId)
     }
   }
@@ -372,7 +374,7 @@ export const runFraudDetection = async ({
   // needs is to iterate over the accepted measurements once.
   const iterateAcceptedMeasurements = function * () {
     for (const m of measurements) {
-      if (m.fraudAssessment !== 'OK') continue
+      if (m.taskingEvaluation !== 'OK') continue
       yield m
     }
   }

lib/platform-stats.js

@@ -67,7 +67,7 @@ export const updateStationsAndParticipants = async (
     }
 
     stationStats.total++
-    if (m.fraudAssessment === 'OK') stationStats.accepted++
+    if (m.taskingEvaluation === 'OK' && m.majorityEvaluation === 'OK') stationStats.accepted++
 
     let subnetsSet = subnets.get(participantId)
     if (!subnetsSet) {

lib/preprocess.js

@@ -22,8 +22,10 @@ export class Measurement {
     // Note: providerId is recorded by spark-publish but we don't use it for evaluations yet
     this.providerId = pointerize(m.provider_id)
     this.spark_version = pointerize(m.spark_version)
-    /** @type {import('./typings.js').FraudAssesment} */
-    this.fraudAssessment = null
+    /** @type {import('./typings.js').TaskingEvaluation} */
+    this.taskingEvaluation = null
+    /** @type {import('./typings.js').MajorityEvaluation} */
+    this.majorityEvaluation = null
     this.inet_group = pointerize(m.inet_group)
     this.finished_at = parseDateTime(m.finished_at)
     this.provider_address = pointerize(m.provider_address)

lib/retrieval-stats.js

@@ -81,8 +81,9 @@ export const buildRetrievalStats = (measurements, telemetryPoint) => {
     const endAt = m.end_at
     const ttfb = startAt && firstByteAt && (firstByteAt - startAt)
     const duration = startAt && endAt && (endAt - startAt)
+    const isAccepted = m.taskingEvaluation === 'OK' && m.majorityEvaluation === 'OK'
 
-    debug('size=%s ttfb=%s duration=%s status=%s valid? %s', byteLength, ttfb, duration, m.status_code, m.fraudAssessment === 'OK')
+    debug('size=%s ttfb=%s duration=%s status=%s accepted? %s', byteLength, ttfb, duration, m.status_code, isAccepted)
     if (byteLength !== undefined && m.status_code === 200) {
       downloadBandwidth += byteLength
       sizeValues.push(byteLength)
@@ -93,7 +94,7 @@ export const buildRetrievalStats = (measurements, telemetryPoint) => {
     const node = `${m.inet_group}::${m.participantAddress}`
     tasksPerNode.set(node, (tasksPerNode.get(node) ?? 0) + 1)
 
-    if (m.fraudAssessment === 'OK') {
+    if (isAccepted) {
       acceptedMeasurementsPerInetGroup.set(m.inet_group, (acceptedMeasurementsPerInetGroup.get(m.inet_group) ?? 0) + 1)
     }
 
@@ -222,7 +223,7 @@ export const recordCommitteeSizes = (committees, point) => {
       data.nodes.add(`${m.inet_group}::${m.participantAddress}`)
       data.measurements++
 
-      if (m.fraudAssessment === 'OK') {
+      if (m.taskingEvaluation === 'OK' && m.majorityEvaluation === 'OK') {
         data.majoritySize = (data.majoritySize ?? 0) + 1
       }
     }

lib/round.js

@@ -1,3 +1,5 @@
+/** @import { Measurement } from './preprocess.js' */
+
 export class RoundData {
   /** @type {Map<string, string>} */
   #knownStrings
@@ -7,8 +9,9 @@ export class RoundData {
    */
   constructor (index) {
     this.index = index
-    /** @type {import('./preprocess.js').Measurement[]} */
+    /** @type {string[]} */
     this.measurementBatches = []
+    /** @type {Measurement[]} */
     this.measurements = []
     this.details = null
     this.#knownStrings = new Map()

lib/typings.d.ts

@@ -34,15 +34,17 @@ export type CommitteeCheckError =
 | 'MAJORITY_NOT_FOUND'
 | 'MINORITY_RESULT'
 
+export type MajorityEvaluation =
+  | 'OK'
+  | CommitteeCheckError
+
 // When adding a new enum value, remember to update the summary initializer inside `evaluate()`
-export type FraudAssesment =
+export type TaskingEvaluation =
   | 'OK'
   | 'TASK_NOT_IN_ROUND'
   | 'TASK_WRONG_NODE'
   | 'DUP_INET_GROUP'
   | 'TOO_MANY_TASKS'
-  | CommitteeCheckError
-
 
 // When adding a new enum value, remember to update the summary initializer inside `reportRetrievalStats()`
 export type RetrievalResult =