feat: meter and settle CDN bandwidth per shared rail

db/migrations/0028_add_cdn_rail_id.sql

@@ -0,0 +1,20 @@
+-- Share the CDN bandwidth rail across a payer's data sets.
+-- Bandwidth is now metered and settled per cdn_rail_id, while cache-miss stays per data set.
+-- cdn_rail_id and cache_miss_rail_id come from the FWSS DataSetCreated event.
+ALTER TABLE data_sets ADD COLUMN cdn_rail_id TEXT;
+ALTER TABLE data_sets ADD COLUMN cache_miss_rail_id TEXT;
+
+-- Bandwidth settlement watermark, keyed by the shared rail rather than the data set.
+-- Cache-miss keeps using data_sets.cdn_payments_settled_until.
+CREATE TABLE cdn_rail_settlement_state (
+  cdn_rail_id TEXT PRIMARY KEY,
+  cdn_payments_settled_until TIMESTAMP WITH TIME ZONE DEFAULT '1970-01-01T00:00:00.000Z' NOT NULL
+);
+
+-- Backfill the bandwidth watermark from existing per-data-set state. Ungrouped data sets
+-- have a unique cdn_rail_id each, so the max collapses to the existing value per rail.
+INSERT INTO cdn_rail_settlement_state (cdn_rail_id, cdn_payments_settled_until)
+SELECT cdn_rail_id, MAX(cdn_payments_settled_until)
+FROM data_sets
+WHERE cdn_rail_id IS NOT NULL
+GROUP BY cdn_rail_id;

indexer/lib/fwss-handlers.js

@@ -48,13 +48,17 @@ export async function handleFWSSDataSetCreated(
         service_provider_id,
         payer_address,
         with_cdn,
-        with_ipfs_indexing
+        with_ipfs_indexing,
+        cdn_rail_id,
+        cache_miss_rail_id
       )
-      VALUES (?, ?, ?, ?, ?)
+      VALUES (?, ?, ?, ?, ?, ?, ?)
       ON CONFLICT DO UPDATE SET
         service_provider_id = excluded.service_provider_id,
         payer_address = excluded.payer_address,
-        with_ipfs_indexing = excluded.with_ipfs_indexing
+        with_ipfs_indexing = excluded.with_ipfs_indexing,
+        cdn_rail_id = excluded.cdn_rail_id,
+        cache_miss_rail_id = excluded.cache_miss_rail_id
     `,
   )
     .bind(
@@ -63,6 +67,10 @@ export async function handleFWSSDataSetCreated(
       payload.payer.toLowerCase(),
       withCDN,
       withIPFSIndexing,
+      payload.cdn_rail_id != null ? String(payload.cdn_rail_id) : null,
+      payload.cache_miss_rail_id != null
+        ? String(payload.cache_miss_rail_id)
+        : null,
     )
     .run()
 }

indexer/test/fwss-cdn-payment-rails.test.js

@@ -366,6 +366,8 @@ describe('webhook ordering scenarios', () => {
       usage_reported_until: '1970-01-01T00:00:00.000Z',
       cdn_payments_settled_until: '1970-01-01T00:00:00.000Z',
       pending_usage_report_tx_hash: null,
+      cdn_rail_id: null,
+      cache_miss_rail_id: null,
     })
 
     // Verify quotas remain unchanged in the egress quotas table
@@ -469,6 +471,8 @@ describe('webhook ordering scenarios', () => {
       usage_reported_until: '1970-01-01T00:00:00.000Z',
       cdn_payments_settled_until: '1970-01-01T00:00:00.000Z',
       pending_usage_report_tx_hash: null,
+      cdn_rail_id: null,
+      cache_miss_rail_id: null,
     })
 
     // Verify accumulated quotas preserved in egress quotas table
@@ -529,6 +533,8 @@ describe('webhook ordering scenarios', () => {
       usage_reported_until: '1970-01-01T00:00:00.000Z',
       cdn_payments_settled_until: '1970-01-01T00:00:00.000Z',
       pending_usage_report_tx_hash: null,
+      cdn_rail_id: null,
+      cache_miss_rail_id: null,
     })
 
     // Verify no quotas exist yet
@@ -586,6 +592,8 @@ describe('webhook ordering scenarios', () => {
       usage_reported_until: '1970-01-01T00:00:00.000Z',
       cdn_payments_settled_until: '1970-01-01T00:00:00.000Z',
       pending_usage_report_tx_hash: null,
+      cdn_rail_id: null,
+      cache_miss_rail_id: null,
     })
   })
 })

indexer/test/indexer.test.js

@@ -82,6 +82,8 @@ describe('piece-retriever.indexer', () => {
           data_set_id: dataSetId,
           payer: '0xPayerAddress',
           provider_id: providerId,
+          cdn_rail_id: '42',
+          cache_miss_rail_id: '43',
           metadata_keys: ['withCDN'],
           metadata_values: [''],
         }),
@@ -111,6 +113,8 @@ describe('piece-retriever.indexer', () => {
       expect(dataSets[0].service_provider_id).toBe(providerId)
       expect(dataSets[0].payer_address).toBe('0xPayerAddress'.toLowerCase())
       expect(dataSets[0].with_cdn).toBe(1)
+      expect(dataSets[0].cdn_rail_id).toBe('42')
+      expect(dataSets[0].cache_miss_rail_id).toBe('43')
 
       expect(walletDetails.length).toBe(1)
       expect(walletDetails[0].is_sanctioned).toBe(0)

payment-settler/bin/payment-settler.js

@@ -1,7 +1,9 @@
 import { getChainClient as defaultGetChainClient } from '../lib/chain.js'
 import {
   getDataSetsForSettlement,
-  settleCDNPaymentRails,
+  getCDNRailsForSettlement,
+  settleCacheMissPaymentRails,
+  settleCDNBandwidthRails,
 } from '../lib/rail-settlement.js'
 import { TransactionMonitorWorkflow } from '@filbeam/workflows'
 import {
@@ -13,7 +15,8 @@ import {
  * @typedef {{
  *   type: 'transaction-retry'
  *   transactionHash: `0x${string}`
- *   dataSetIds: string[]
+ *   settlementType: 'cache-miss' | 'bandwidth'
+ *   ids: string[]
  * }} TransactionRetryMessage
  */
 
@@ -22,10 +25,64 @@ import {
  *   type: 'settlement-confirmed'
  *   transactionHash: `0x${string}`
  *   blockNumber: string
- *   dataSetIds: string[]
+ *   settlementType: 'cache-miss' | 'bandwidth'
+ *   ids: string[]
  * }} SettlementConfirmedMessage
  */
 
+/**
+ * Splits ids into batches, settles each batch, and returns metadata for the
+ * batches that succeeded.
+ *
+ * @param {object} args
+ * @param {Env} args.env
+ * @param {string[]} args.ids
+ * @param {(batch: string[], batchId: string) => Promise<`0x${string}`>} args.settle
+ * @param {'cache-miss' | 'bandwidth'} args.settlementType
+ * @returns {Promise<
+ *   {
+ *     transactionHash: `0x${string}`
+ *     settlementType: 'cache-miss' | 'bandwidth'
+ *     ids: string[]
+ *   }[]
+ * >}
+ */
+async function settleInBatches({ env, ids, settle, settlementType }) {
+  if (ids.length === 0) return []
+
+  const batches = []
+  for (let i = 0; i < ids.length; i += env.SETTLEMENT_BATCH_SIZE) {
+    batches.push(ids.slice(i, i + env.SETTLEMENT_BATCH_SIZE))
+  }
+
+  console.log(
+    `Prepared ${batches.length} ${settlementType} batches for settlement`,
+  )
+
+  const results = await Promise.allSettled(
+    batches.map((batch, ix) => settle(batch, `${settlementType}-batch-${ix}`)),
+  )
+
+  const successfulBatches = []
+  for (let i = 0; i < results.length; i++) {
+    const result = results[i]
+    if (result.status === 'fulfilled') {
+      successfulBatches.push({
+        transactionHash: result.value,
+        settlementType,
+        ids: batches[i],
+      })
+    } else {
+      console.error(
+        `Failed to settle ${settlementType} batch ${i + 1} (${batches[i].join(', ')}):`,
+        result.reason,
+      )
+    }
+  }
+
+  return successfulBatches
+}
+
 export default {
   /**
    * @param {any} _controller
@@ -42,74 +99,70 @@ export default {
     console.log('Starting rail settlement worker')
 
     try {
-      const dataSetIds = await getDataSetsForSettlement(env.DB)
+      const chainClient = getChainClient(env)
 
-      if (dataSetIds.length === 0) {
+      const [dataSetIds, cdnRailIds] = await Promise.all([
+        getDataSetsForSettlement(env.DB),
+        getCDNRailsForSettlement(env.DB),
+      ])
+
+      if (dataSetIds.length === 0 && cdnRailIds.length === 0) {
         console.log('No active data sets found for settlement')
         return
       }
 
       console.log(
-        `Found ${dataSetIds.length} data sets for settlement:`,
-        dataSetIds,
+        `Found ${dataSetIds.length} data sets and ${cdnRailIds.length} bandwidth rails for settlement`,
       )
 
-      const batches = []
-      for (let i = 0; i < dataSetIds.length; i += env.SETTLEMENT_BATCH_SIZE) {
-        batches.push(dataSetIds.slice(i, i + env.SETTLEMENT_BATCH_SIZE))
-      }
-
-      console.log(`Prepared ${batches.length} batches for settlement`)
+      const [cacheMissBatches, bandwidthBatches] = await Promise.all([
+        // Cache-miss settlement: once per data set.
+        settleInBatches({
+          env,
+          ids: dataSetIds,
+          settle: (batch, batchId) =>
+            settleCacheMissPaymentRails({
+              env,
+              batchId,
+              dataSetIds: batch,
+              ...chainClient,
+            }),
+          settlementType: 'cache-miss',
+        }),
+        // Bandwidth settlement: once per shared cdn_rail_id.
+        settleInBatches({
+          env,
+          ids: cdnRailIds,
+          settle: (batch, batchId) =>
+            settleCDNBandwidthRails({
+              env,
+              batchId,
+              cdnRailIds: batch,
+              ...chainClient,
+            }),
+          settlementType: 'bandwidth',
+        }),
+      ])
 
-      const chainClient = getChainClient(env)
-
-      const results = await Promise.allSettled(
-        batches.map((batch, ix) =>
-          settleCDNPaymentRails({
-            env,
-            batchId: `batch-${ix}`,
-            dataSetIds: batch,
-            ...chainClient,
-          }),
-        ),
-      )
+      const workflows = [...cacheMissBatches, ...bandwidthBatches]
 
-      /** @type {{ transactionHash: `0x${string}`; dataSetIds: string[] }[]} */
-      const successfulBatches = []
-      for (let i = 0; i < results.length; i++) {
-        const result = results[i]
-        if (result.status === 'fulfilled') {
-          successfulBatches.push({
-            transactionHash: result.value,
-            dataSetIds: batches[i],
-          })
-        } else {
-          console.error(
-            `Failed to settle batch ${i + 1} (data sets: ${batches[i].join(', ')}):`,
-            result.reason,
-          )
-        }
-      }
-
-      if (successfulBatches.length > 0) {
+      if (workflows.length > 0) {
         await env.TRANSACTION_MONITOR_WORKFLOW.createBatch(
-          successfulBatches.map(({ transactionHash, dataSetIds }) => ({
+          workflows.map(({ transactionHash, settlementType, ids }) => ({
             id: `payment-settler-${transactionHash}-${Date.now()}`,
             params: {
               transactionHash,
               metadata: {
                 onSuccess: 'settlement-confirmed',
-                successData: { dataSetIds },
-                retryData: { dataSetIds },
+                successData: { settlementType, ids },
+                retryData: { settlementType, ids },
               },
             },
           })),
         )
       }
 
-      console.log(
-        `Settled ${successfulBatches.length} of ${batches.length} batches`,
-      )
+      console.log(`Created ${workflows.length} settlement monitor workflows`)
     } catch (error) {
       console.error('Settlement process failed:', error)
       throw error

payment-settler/lib/FilBeamOperator.abi.json

@@ -1,7 +1,7 @@
 [
   {
     "type": "function",
-    "name": "settleCDNPaymentRails",
+    "name": "settleCacheMissPaymentRails",
     "inputs": [
       {
         "name": "dataSetIds",
@@ -11,5 +11,18 @@
     ],
     "outputs": [],
     "stateMutability": "nonpayable"
+  },
+  {
+    "type": "function",
+    "name": "settleCDNBandwidthRails",
+    "inputs": [
+      {
+        "name": "cdnRailIds",
+        "type": "uint256[]",
+        "internalType": "uint256[]"
+      }
+    ],
+    "outputs": [],
+    "stateMutability": "nonpayable"
   }
 ]