diff --git a/db/migrations/0028_add_cdn_rail_id.sql b/db/migrations/0028_add_cdn_rail_id.sql
new file mode 100644
index 00000000..88745c98
--- /dev/null
+++ b/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;
diff --git a/indexer/lib/fwss-handlers.js b/indexer/lib/fwss-handlers.js
index 0ce79bae..75a39724 100644
--- a/indexer/lib/fwss-handlers.js
+++ b/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()
 }
diff --git a/indexer/test/fwss-cdn-payment-rails.test.js b/indexer/test/fwss-cdn-payment-rails.test.js
index 0d23590a..77aead85 100644
--- a/indexer/test/fwss-cdn-payment-rails.test.js
+++ b/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,
     })
   })
 })
diff --git a/indexer/test/indexer.test.js b/indexer/test/indexer.test.js
index 94b8aacf..72aa98c2 100644
--- a/indexer/test/indexer.test.js
+++ b/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)
diff --git a/payment-settler/bin/payment-settler.js b/payment-settler/bin/payment-settler.js
index 4b6ea4fb..0a0ab9a5 100644
--- a/payment-settler/bin/payment-settler.js
+++ b/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
diff --git a/payment-settler/lib/FilBeamOperator.abi.json b/payment-settler/lib/FilBeamOperator.abi.json
index dcd8dd42..dfd5a168 100644
--- a/payment-settler/lib/FilBeamOperator.abi.json
+++ b/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"
   }
 ]
diff --git a/payment-settler/lib/queue-handlers.js b/payment-settler/lib/queue-handlers.js
index 011926f1..341aecd1 100644
--- a/payment-settler/lib/queue-handlers.js
+++ b/payment-settler/lib/queue-handlers.js
@@ -7,7 +7,8 @@ import { epochToTimestampMs } from './epoch.js'
  * @typedef {{
  *   type: 'transaction-retry'
  *   transactionHash: `0x${string}`
- *   dataSetIds: string[]
+ *   settlementType: 'cache-miss' | 'bandwidth'
+ *   ids: string[]
  * }} TransactionRetryMessage
  */
 
@@ -16,21 +17,27 @@ import { epochToTimestampMs } from './epoch.js'
  *   type: 'settlement-confirmed'
  *   transactionHash: `0x${string}`
  *   blockNumber: string
- *   dataSetIds: string[]
+ *   settlementType: 'cache-miss' | 'bandwidth'
+ *   ids: string[]
  * }} SettlementConfirmedMessage
  */
 
 /**
- * Handles settlement confirmed queue messages
+ * Handles settlement confirmed queue messages.
+ *
+ * Cache-miss settlement advances the per-data-set watermark
+ * (data_sets.cdn_payments_settled_until), bandwidth settlement advances the
+ * per-rail watermark (cdn_rail_settlement_state.cdn_payments_settled_until).
  *
  * @param {SettlementConfirmedMessage} message
  * @param {Env} env
  */
 export async function handleSettlementConfirmedQueueMessage(message, env) {
-  const { transactionHash, blockNumber, dataSetIds } = message
+  const { transactionHash, blockNumber, settlementType, ids } = message
   assert(transactionHash, 'transactionHash is required')
   assert(blockNumber, 'blockNumber is required')
-  assert(dataSetIds, 'dataSetIds is required')
+  assert(settlementType, 'settlementType is required')
+  assert(ids, 'ids is required')
 
   console.log(`Processing settlement confirmation for hash: ${transactionHash}`)
 
@@ -42,21 +49,43 @@ export async function handleSettlementConfirmedQueueMessage(message, env) {
       ),
     ).toISOString()
 
-    const placeholders = dataSetIds.map(() => '?').join(', ')
-    await env.DB.prepare(
-      `
-      UPDATE data_sets
-      SET cdn_payments_settled_until = ?
-      WHERE id IN (${placeholders})
-        AND cdn_payments_settled_until < ?
-      `,
-    )
-      .bind(settledUntil, ...dataSetIds, settledUntil)
-      .run()
+    if (settlementType === 'bandwidth') {
+      const valuePlaceholders = ids.map(() => '(?, ?)').join(', ')
+      const bindings = ids.flatMap((id) => [id, settledUntil])
+      await env.DB.prepare(
+        `
+        INSERT INTO cdn_rail_settlement_state (cdn_rail_id, cdn_payments_settled_until)
+        VALUES ${valuePlaceholders}
+        ON CONFLICT (cdn_rail_id) DO UPDATE SET
+          cdn_payments_settled_until = MAX(
+            cdn_rail_settlement_state.cdn_payments_settled_until,
+            excluded.cdn_payments_settled_until
+          )
+        `,
+      )
+        .bind(...bindings)
+        .run()
 
-    console.log(
-      `Updated cdn_payments_settled_until to ${settledUntil} for ${dataSetIds.length} data sets`,
-    )
+      console.log(
+        `Updated bandwidth cdn_payments_settled_until to ${settledUntil} for ${ids.length} rails`,
+      )
+    } else {
+      const placeholders = ids.map(() => '?').join(', ')
+      await env.DB.prepare(
+        `
+        UPDATE data_sets
+        SET cdn_payments_settled_until = ?
+        WHERE id IN (${placeholders})
+          AND cdn_payments_settled_until < ?
+        `,
+      )
+        .bind(settledUntil, ...ids, settledUntil)
+        .run()
+
+      console.log(
+        `Updated cache-miss cdn_payments_settled_until to ${settledUntil} for ${ids.length} data sets`,
+      )
+    }
   } catch (error) {
     console.error(
       `Failed to process settlement confirmation for hash: ${transactionHash}`,
@@ -80,9 +109,10 @@ export async function handleTransactionRetryQueueMessage(
     getRecentSendMessage = defaultGetRecentSendMessage,
   } = {},
 ) {
-  const { transactionHash, dataSetIds } = message
+  const { transactionHash, settlementType, ids } = message
   assert(transactionHash, 'transactionHash is required')
-  assert(dataSetIds, 'dataSetIds is required')
+  assert(settlementType, 'settlementType is required')
+  assert(ids, 'ids is required')
 
   console.log(`Processing transaction retry for hash: ${transactionHash}`)
 
@@ -105,7 +135,8 @@ export async function handleTransactionRetryQueueMessage(
           type: 'settlement-confirmed',
           transactionHash,
           blockNumber: receipt.blockNumber.toString(),
-          dataSetIds,
+          settlementType,
+          ids,
         })
         console.log(
           `Sent confirmation message to queue for already confirmed transaction ${transactionHash}`,
@@ -179,8 +210,8 @@ export async function handleTransactionRetryQueueMessage(
         transactionHash: retryHash,
         metadata: {
           onSuccess: 'settlement-confirmed',
-          successData: { dataSetIds },
-          retryData: { dataSetIds },
+          successData: { settlementType, ids },
+          retryData: { settlementType, ids },
         },
       },
     })
diff --git a/payment-settler/lib/rail-settlement.js b/payment-settler/lib/rail-settlement.js
index e434b7ce..d80c91e9 100644
--- a/payment-settler/lib/rail-settlement.js
+++ b/payment-settler/lib/rail-settlement.js
@@ -2,7 +2,7 @@
 import filbeamAbi from '../lib/FilBeamOperator.abi.json'
 
 /**
- * Fetches data sets that need CDN payment rail settlement
+ * Fetches data sets that need cache-miss payment rail settlement.
  *
  * Only settle data sets with recently reported usage (within last 30 days).
  * This prevents unnecessary settlement attempts for inactive or abandoned data
@@ -33,6 +33,38 @@ export async function getDataSetsForSettlement(db) {
 }
 
 /**
+ * Fetches the distinct shared bandwidth rails that need settlement.
+ *
+ * A rail is eligible when at least one of its data sets is eligible for
+ * settlement (same criteria as {@link getDataSetsForSettlement}). Because data
+ * sets in a CDN group share one cdn_rail_id, the bandwidth is settled once per
+ * rail rather than once per data set.
+ *
+ * @param {D1Database} db - The database connection
+ * @returns {Promise<string[]>} Array of cdn_rail_id values that need settlement
+ */
+export async function getCDNRailsForSettlement(db) {
+  const result = await db
+    .prepare(
+      `
+      SELECT DISTINCT data_sets.cdn_rail_id
+      FROM data_sets
+      LEFT JOIN wallet_details ON data_sets.payer_address = wallet_details.address
+      WHERE (data_sets.with_cdn = 1 OR data_sets.lockup_unlocks_at >= datetime('now'))
+        AND data_sets.terminate_service_tx_hash IS NULL
+        AND data_sets.usage_reported_until >= datetime('now', '-30 days')
+        AND data_sets.cdn_rail_id IS NOT NULL
+        AND (wallet_details.is_sanctioned IS NULL OR wallet_details.is_sanctioned = 0)
+      `,
+    )
+    .all()
+
+  return result.results.map((row) => String(row.cdn_rail_id))
+}
+
+/**
+ * Settles cache-miss payment rails for a batch of data sets.
+ *
  * @param {object} args
  * @param {Env} args.env
  * @param {string} args.batchId
@@ -42,7 +74,7 @@ export async function getDataSetsForSettlement(db) {
  * @param {string[]} args.dataSetIds
  * @returns {Promise<`0x${string}`>}
  */
-export async function settleCDNPaymentRails({
+export async function settleCacheMissPaymentRails({
   env,
   batchId,
   publicClient,
@@ -50,17 +82,56 @@ export async function settleCDNPaymentRails({
   account,
   dataSetIds,
 }) {
-  console.log(`[${batchId}] Settling ${dataSetIds.length} data sets...`)
+  console.log(
+    `[${batchId}] Settling cache-miss for ${dataSetIds.length} data sets...`,
+  )
 
   const { request } = await publicClient.simulateContract({
     account,
     abi: filbeamAbi,
     address: env.FILBEAM_OPERATOR_CONTRACT_ADDRESS,
-    functionName: 'settleCDNPaymentRails',
+    functionName: 'settleCacheMissPaymentRails',
     args: [dataSetIds.map((id) => BigInt(id))],
   })
 
   const txHash = await walletClient.writeContract(request)
-  console.log(`[${batchId}] Settlement transaction sent: ${txHash}`)
+  console.log(`[${batchId}] Cache-miss settlement transaction sent: ${txHash}`)
+  return txHash
+}
+
+/**
+ * Settles the shared bandwidth rails for a batch of cdn_rail_ids.
+ *
+ * @param {object} args
+ * @param {Env} args.env
+ * @param {string} args.batchId
+ * @param {PublicClient} args.publicClient
+ * @param {WalletClient} args.walletClient
+ * @param {PrivateKeyAccount} args.account
+ * @param {string[]} args.cdnRailIds
+ * @returns {Promise<`0x${string}`>}
+ */
+export async function settleCDNBandwidthRails({
+  env,
+  batchId,
+  publicClient,
+  walletClient,
+  account,
+  cdnRailIds,
+}) {
+  console.log(
+    `[${batchId}] Settling bandwidth for ${cdnRailIds.length} rails...`,
+  )
+
+  const { request } = await publicClient.simulateContract({
+    account,
+    abi: filbeamAbi,
+    address: env.FILBEAM_OPERATOR_CONTRACT_ADDRESS,
+    functionName: 'settleCDNBandwidthRails',
+    args: [cdnRailIds.map((id) => BigInt(id))],
+  })
+
+  const txHash = await walletClient.writeContract(request)
+  console.log(`[${batchId}] Bandwidth settlement transaction sent: ${txHash}`)
   return txHash
 }
diff --git a/payment-settler/test/queue-handlers.test.js b/payment-settler/test/queue-handlers.test.js
index c8489026..aef89acd 100644
--- a/payment-settler/test/queue-handlers.test.js
+++ b/payment-settler/test/queue-handlers.test.js
@@ -22,20 +22,22 @@ describe('handleSettlementConfirmedQueueMessage', () => {
     vi.useFakeTimers()
     vi.setSystemTime(date)
     await env.DB.exec('DELETE FROM data_sets')
+    await env.DB.exec('DELETE FROM cdn_rail_settlement_state')
   })
 
   afterEach(() => {
     vi.useRealTimers()
   })
 
-  it('updates single dataset', async () => {
+  it('updates cache-miss watermark for a single dataset', async () => {
     const dataSetId = nextId()
     const transactionHash = '0xTest'
     const message = {
       type: 'settlement-confirmed',
       transactionHash,
       blockNumber: BLOCK_NUMBER_2000,
-      dataSetIds: [dataSetId],
+      settlementType: 'cache-miss',
+      ids: [dataSetId],
     }
 
     await withDataSet(env, {
@@ -57,17 +59,18 @@ describe('handleSettlementConfirmedQueueMessage', () => {
     })
   })
 
-  it('updates multiple datasets', async () => {
+  it('updates cache-miss watermark for multiple datasets', async () => {
     const transactionHash = '0xTest'
-    const dataSetIds = [nextId(), nextId(), nextId()]
+    const ids = [nextId(), nextId(), nextId()]
     const message = {
       type: 'settlement-confirmed',
       transactionHash,
       blockNumber: BLOCK_NUMBER_2000,
-      dataSetIds,
+      settlementType: 'cache-miss',
+      ids,
     }
 
-    for (const id of dataSetIds) {
+    for (const id of ids) {
       await withDataSet(env, {
         id,
         withCDN: true,
@@ -82,21 +85,80 @@ describe('handleSettlementConfirmedQueueMessage', () => {
     ).all()
 
     expect(results).toStrictEqual(
-      dataSetIds.sort().map((id) => ({
+      ids.sort().map((id) => ({
         id,
         cdn_payments_settled_until: TIMESTAMP_AT_BLOCK_2000,
       })),
     )
   })
 
-  it('does not regress if timestamp is already newer (idempotency)', async () => {
+  it('updates bandwidth watermark per cdn_rail_id', async () => {
+    const transactionHash = '0xTest'
+    const cdnRailIds = ['rail-a', 'rail-b']
+    const message = {
+      type: 'settlement-confirmed',
+      transactionHash,
+      blockNumber: BLOCK_NUMBER_2000,
+      settlementType: 'bandwidth',
+      ids: cdnRailIds,
+    }
+
+    await handleSettlementConfirmedQueueMessage(message, testEnv)
+
+    const { results } = await env.DB.prepare(
+      'SELECT cdn_rail_id, cdn_payments_settled_until FROM cdn_rail_settlement_state ORDER BY cdn_rail_id',
+    ).all()
+
+    expect(results).toStrictEqual([
+      {
+        cdn_rail_id: 'rail-a',
+        cdn_payments_settled_until: TIMESTAMP_AT_BLOCK_2000,
+      },
+      {
+        cdn_rail_id: 'rail-b',
+        cdn_payments_settled_until: TIMESTAMP_AT_BLOCK_2000,
+      },
+    ])
+  })
+
+  it('does not regress the bandwidth watermark (idempotency)', async () => {
+    const transactionHash = '0xTest'
+    const message = {
+      type: 'settlement-confirmed',
+      transactionHash,
+      blockNumber: BLOCK_NUMBER_2000,
+      settlementType: 'bandwidth',
+      ids: ['rail-a'],
+    }
+
+    await env.DB.prepare(
+      'INSERT INTO cdn_rail_settlement_state (cdn_rail_id, cdn_payments_settled_until) VALUES (?, ?)',
+    )
+      .bind('rail-a', TIMESTAMP_AT_BLOCK_3000)
+      .run()
+
+    await handleSettlementConfirmedQueueMessage(message, testEnv)
+
+    const railState = await env.DB.prepare(
+      'SELECT cdn_payments_settled_until FROM cdn_rail_settlement_state WHERE cdn_rail_id = ?',
+    )
+      .bind('rail-a')
+      .first()
+
+    expect(railState).toStrictEqual({
+      cdn_payments_settled_until: TIMESTAMP_AT_BLOCK_3000,
+    })
+  })
+
+  it('does not regress the cache-miss watermark (idempotency)', async () => {
     const dataSetId = nextId()
     const transactionHash = '0xTest'
     const message = {
       type: 'settlement-confirmed',
       transactionHash,
       blockNumber: BLOCK_NUMBER_2000,
-      dataSetIds: [dataSetId],
+      settlementType: 'cache-miss',
+      ids: [dataSetId],
     }
 
     await withDataSet(env, {
@@ -133,7 +195,8 @@ describe('handleSettlementConfirmedQueueMessage', () => {
       type: 'settlement-confirmed',
       transactionHash,
       blockNumber: BLOCK_NUMBER_2000,
-      dataSetIds: [dataSetId],
+      settlementType: 'cache-miss',
+      ids: [dataSetId],
     }
 
     await withDataSet(env, {
@@ -161,7 +224,8 @@ describe('handleSettlementConfirmedQueueMessage', () => {
     const message = {
       type: 'settlement-confirmed',
       blockNumber: BLOCK_NUMBER_2000,
-      dataSetIds: ['1'],
+      settlementType: 'cache-miss',
+      ids: ['1'],
     }
 
     await expect(
@@ -173,7 +237,8 @@ describe('handleSettlementConfirmedQueueMessage', () => {
     const message = {
       type: 'settlement-confirmed',
       transactionHash: '0xTest',
-      dataSetIds: ['1'],
+      settlementType: 'cache-miss',
+      ids: ['1'],
     }
 
     await expect(
@@ -181,16 +246,17 @@ describe('handleSettlementConfirmedQueueMessage', () => {
     ).rejects.toThrow('blockNumber')
   })
 
-  it('throws on missing dataSetIds', async () => {
+  it('throws on missing ids', async () => {
     const message = {
       type: 'settlement-confirmed',
       transactionHash: '0xTest',
       blockNumber: BLOCK_NUMBER_2000,
+      settlementType: 'cache-miss',
     }
 
     await expect(
       handleSettlementConfirmedQueueMessage(message, testEnv),
-    ).rejects.toThrow('dataSetIds')
+    ).rejects.toThrow('ids')
   })
 })
 
@@ -254,12 +320,13 @@ describe('handleTransactionRetryQueueMessage', () => {
   })
 
   it('sends settlement-confirmed message when original TX is already confirmed', async () => {
-    const dataSetIds = ['1', '2']
+    const ids = ['1', '2']
     const transactionHash = '0xOriginalHash'
     const message = {
       type: 'transaction-retry',
       transactionHash,
-      dataSetIds,
+      settlementType: 'cache-miss',
+      ids,
     }
 
     const { mock: mockGetChainClient } = createMockGetChainClient({
@@ -284,17 +351,19 @@ describe('handleTransactionRetryQueueMessage', () => {
       type: 'settlement-confirmed',
       transactionHash,
       blockNumber: '12345',
-      dataSetIds,
+      settlementType: 'cache-miss',
+      ids,
     })
   })
 
   it('passes metadata to new workflow when retrying', async () => {
-    const dataSetIds = ['1', '2']
+    const ids = ['1', '2']
     const transactionHash = '0xOriginalHash'
     const message = {
       type: 'transaction-retry',
       transactionHash,
-      dataSetIds,
+      settlementType: 'bandwidth',
+      ids,
     }
 
     const { mock: mockGetChainClient } = createMockGetChainClient({
@@ -321,8 +390,8 @@ describe('handleTransactionRetryQueueMessage', () => {
         transactionHash: '0xNewRetryHash',
         metadata: {
           onSuccess: 'settlement-confirmed',
-          successData: { dataSetIds },
-          retryData: { dataSetIds },
+          successData: { settlementType: 'bandwidth', ids },
+          retryData: { settlementType: 'bandwidth', ids },
         },
       },
     })
diff --git a/payment-settler/test/rail-settlement.test.js b/payment-settler/test/rail-settlement.test.js
index 7b1e0bcb..347f73ee 100644
--- a/payment-settler/test/rail-settlement.test.js
+++ b/payment-settler/test/rail-settlement.test.js
@@ -1,6 +1,9 @@
 import { describe, it, expect, beforeEach } from 'vitest'
 import { env } from 'cloudflare:test'
-import { getDataSetsForSettlement } from '../lib/rail-settlement.js'
+import {
+  getDataSetsForSettlement,
+  getCDNRailsForSettlement,
+} from '../lib/rail-settlement.js'
 import {
   withDataSet,
   withWallet,
@@ -564,4 +567,80 @@ describe('rail settlement', () => {
       expect(dataSetIds).toStrictEqual([id2])
     })
   })
+
+  describe('getCDNRailsForSettlement', () => {
+    beforeEach(async () => {
+      await env.DB.prepare('DELETE FROM data_sets').run()
+      await env.DB.prepare('DELETE FROM wallet_details').run()
+    })
+
+    it('returns one rail for data sets that share a cdn_rail_id', async () => {
+      const id1 = nextId()
+      const id2 = nextId()
+
+      await withDataSet(env, {
+        id: id1,
+        withCDN: true,
+        cdnRailId: 'shared-rail',
+        usageReportedUntil: getDaysAgo(5),
+      })
+      await withDataSet(env, {
+        id: id2,
+        withCDN: true,
+        cdnRailId: 'shared-rail',
+        usageReportedUntil: getDaysAgo(10),
+      })
+
+      const cdnRailIds = await getCDNRailsForSettlement(env.DB)
+
+      expect(cdnRailIds).toStrictEqual(['shared-rail'])
+    })
+
+    it('excludes terminated, sanctioned, inactive and null-rail data sets', async () => {
+      const sanctionedAddress = '0xSanctionedRail'
+      await withWallet(env, sanctionedAddress, true)
+
+      // Eligible
+      await withDataSet(env, {
+        id: nextId(),
+        withCDN: true,
+        cdnRailId: 'rail-eligible',
+        usageReportedUntil: getDaysAgo(5),
+      })
+      // Terminated
+      await withDataSet(env, {
+        id: nextId(),
+        withCDN: true,
+        cdnRailId: 'rail-terminated',
+        terminateServiceTxHash: '0xabc',
+        usageReportedUntil: getDaysAgo(5),
+      })
+      // Sanctioned payer
+      await withDataSet(env, {
+        id: nextId(),
+        withCDN: true,
+        cdnRailId: 'rail-sanctioned',
+        payerAddress: sanctionedAddress,
+        usageReportedUntil: getDaysAgo(5),
+      })
+      // No recent usage
+      await withDataSet(env, {
+        id: nextId(),
+        withCDN: true,
+        cdnRailId: 'rail-stale',
+        usageReportedUntil: getDaysAgo(45),
+      })
+      // Null cdn_rail_id
+      await withDataSet(env, {
+        id: nextId(),
+        withCDN: true,
+        cdnRailId: null,
+        usageReportedUntil: getDaysAgo(5),
+      })
+
+      const cdnRailIds = await getCDNRailsForSettlement(env.DB)
+
+      expect(cdnRailIds).toStrictEqual(['rail-eligible'])
+    })
+  })
 })
diff --git a/payment-settler/test/test-helpers.js b/payment-settler/test/test-helpers.js
index 7a039876..3b913172 100644
--- a/payment-settler/test/test-helpers.js
+++ b/payment-settler/test/test-helpers.js
@@ -18,6 +18,10 @@ export async function withDataSet(
     terminateServiceTxHash = null,
     lockupUnlocksAt = null,
     usageReportedUntil = null,
+    // Rail ids are uint256 on-chain, so keep fixtures numeric (the settler
+    // converts them with BigInt).
+    cdnRailId = String(Number(id) + 1000),
+    cacheMissRailId = String(Number(id) + 2000),
   },
 ) {
   // Ensure service provider exists
@@ -33,7 +37,7 @@ export async function withDataSet(
   const lockupUnlocksAtValue = lockupUnlocksAt
 
   await env.DB.prepare(
-    `INSERT INTO data_sets (id, service_provider_id, payer_address, with_cdn, terminate_service_tx_hash, lockup_unlocks_at, usage_reported_until) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+    `INSERT INTO data_sets (id, service_provider_id, payer_address, with_cdn, terminate_service_tx_hash, lockup_unlocks_at, usage_reported_until, cdn_rail_id, cache_miss_rail_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
   )
     .bind(
       String(id),
@@ -43,6 +47,8 @@ export async function withDataSet(
       terminateServiceTxHash,
       lockupUnlocksAtValue,
       usageReportedUntilValue,
+      cdnRailId,
+      cacheMissRailId,
     )
     .run()
 }
diff --git a/payment-settler/test/worker.test.js b/payment-settler/test/worker.test.js
index 6d49e28e..367904c4 100644
--- a/payment-settler/test/worker.test.js
+++ b/payment-settler/test/worker.test.js
@@ -96,70 +96,97 @@ describe('payment settler scheduled handler', () => {
 
     expect(mockGetChainClient).toHaveBeenCalledWith(mockEnv)
 
-    expect(simulateContractCalls).toStrictEqual([
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-      },
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-      },
-    ])
-
-    expect(writeContractCalls).toStrictEqual([
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-        mockedRequest: true,
-      },
-      {
+    // Two data sets -> two cache-miss batches (by data set) and two bandwidth
+    // batches (by rail), with SETTLEMENT_BATCH_SIZE = 1.
+    const cacheMissCalls = simulateContractCalls.filter(
+      (call) => call.functionName === 'settleCacheMissPaymentRails',
+    )
+    const bandwidthCalls = simulateContractCalls.filter(
+      (call) => call.functionName === 'settleCDNBandwidthRails',
+    )
+    expect(cacheMissCalls).toHaveLength(2)
+    expect(bandwidthCalls).toHaveLength(2)
+    expect(simulateContractCalls).toHaveLength(4)
+    for (const call of simulateContractCalls) {
+      expect(call).toStrictEqual({
         account: mockAccount,
         abi: expect.any(Array),
         address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
+        functionName: expect.any(String),
         args: [[expect.any(BigInt)]],
-        mockedRequest: true,
-      },
-    ])
-
-    expect(mockWorkflow.createBatch).toHaveBeenCalledWith([
-      {
-        id: expect.stringMatching(
-          /^payment-settler-0xMockTransactionHash-\d+$/,
-        ),
-        params: {
-          transactionHash: '0xMockTransactionHash',
-          metadata: {
-            onSuccess: 'settlement-confirmed',
-            successData: { dataSetIds: [id1] },
-            retryData: { dataSetIds: [id1] },
+      })
+    }
+
+    expect(writeContractCalls).toHaveLength(4)
+
+    expect(mockWorkflow.createBatch).toHaveBeenCalledWith(
+      expect.arrayContaining([
+        {
+          id: expect.stringMatching(
+            /^payment-settler-0xMockTransactionHash-\d+$/,
+          ),
+          params: {
+            transactionHash: '0xMockTransactionHash',
+            metadata: {
+              onSuccess: 'settlement-confirmed',
+              successData: { settlementType: 'cache-miss', ids: [id1] },
+              retryData: { settlementType: 'cache-miss', ids: [id1] },
+            },
           },
         },
-      },
-      {
-        id: expect.stringMatching(
-          /^payment-settler-0xMockTransactionHash-\d+$/,
-        ),
-        params: {
-          transactionHash: '0xMockTransactionHash',
-          metadata: {
-            onSuccess: 'settlement-confirmed',
-            successData: { dataSetIds: [id2] },
-            retryData: { dataSetIds: [id2] },
+        {
+          id: expect.stringMatching(
+            /^payment-settler-0xMockTransactionHash-\d+$/,
+          ),
+          params: {
+            transactionHash: '0xMockTransactionHash',
+            metadata: {
+              onSuccess: 'settlement-confirmed',
+              successData: { settlementType: 'cache-miss', ids: [id2] },
+              retryData: { settlementType: 'cache-miss', ids: [id2] },
+            },
+          },
+        },
+        {
+          id: expect.stringMatching(
+            /^payment-settler-0xMockTransactionHash-\d+$/,
+          ),
+          params: {
+            transactionHash: '0xMockTransactionHash',
+            metadata: {
+              onSuccess: 'settlement-confirmed',
+              successData: {
+                settlementType: 'bandwidth',
+                ids: [String(Number(id1) + 1000)],
+              },
+              retryData: {
+                settlementType: 'bandwidth',
+                ids: [String(Number(id1) + 1000)],
+              },
+            },
+          },
+        },
+        {
+          id: expect.stringMatching(
+            /^payment-settler-0xMockTransactionHash-\d+$/,
+          ),
+          params: {
+            transactionHash: '0xMockTransactionHash',
+            metadata: {
+              onSuccess: 'settlement-confirmed',
+              successData: {
+                settlementType: 'bandwidth',
+                ids: [String(Number(id2) + 1000)],
+              },
+              retryData: {
+                settlementType: 'bandwidth',
+                ids: [String(Number(id2) + 1000)],
+              },
+            },
           },
         },
-      },
-    ])
+      ]),
+    )
   })
 
   it('should handle no active data sets gracefully', async () => {
@@ -204,40 +231,15 @@ describe('payment settler scheduled handler', () => {
       { getChainClient: mockGetChainClient },
     )
 
-    expect(simulateContractCalls).toStrictEqual([
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-      },
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-      },
-    ])
-    expect(writeContractCalls).toStrictEqual([
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-        mockedRequest: true,
-      },
-      {
-        account: mockAccount,
-        abi: expect.any(Array),
-        address: '0xTestContractAddress',
-        functionName: 'settleCDNPaymentRails',
-        args: [[expect.any(BigInt)]],
-        mockedRequest: true,
-      },
-    ])
+    const cacheMissCalls = simulateContractCalls.filter(
+      (call) => call.functionName === 'settleCacheMissPaymentRails',
+    )
+    const bandwidthCalls = simulateContractCalls.filter(
+      (call) => call.functionName === 'settleCDNBandwidthRails',
+    )
+    expect(cacheMissCalls).toHaveLength(2)
+    expect(bandwidthCalls).toHaveLength(2)
+    expect(writeContractCalls).toHaveLength(4)
   })
 
   it('should log error and continue when contract simulation fails', async () => {
@@ -293,8 +295,8 @@ describe('payment settler scheduled handler', () => {
       { getChainClient: mockGetChainClient },
     )
 
-    // Simulation was attempted
-    expect(simulateContractCalls).toHaveLength(1)
+    // Simulation was attempted for both cache-miss and bandwidth
+    expect(simulateContractCalls).toHaveLength(2)
     // No workflow should have been created since write failed
     expect(mockWorkflow.createBatch).not.toHaveBeenCalled()
   })
@@ -383,28 +385,17 @@ describe('payment settler scheduled handler', () => {
       { getChainClient: mockGetChainClient },
     )
 
-    // All three simulations were attempted
-    expect(simulateContractCalls).toHaveLength(3)
+    // Three data sets -> three cache-miss and three bandwidth simulations
+    expect(simulateContractCalls).toHaveLength(6)
 
-    // Only two successful batches were written
-    expect(writeContractCalls).toHaveLength(2)
+    // One failed simulation -> the other five batches were still written
+    expect(writeContractCalls).toHaveLength(5)
 
-    // Only two successful transactions were monitored
-    expect(mockWorkflow.createBatch).toHaveBeenCalledWith([
-      {
-        id: expect.stringMatching(
-          /^payment-settler-0xMockTransactionHash-\d+$/,
-        ),
-        params: {
-          transactionHash: '0xMockTransactionHash',
-          metadata: {
-            onSuccess: 'settlement-confirmed',
-            successData: { dataSetIds: [id1] },
-            retryData: { dataSetIds: [id1] },
-          },
-        },
-      },
-      {
+    // Only the successful batches were monitored
+    const [workflows] = mockWorkflow.createBatch.mock.calls[0]
+    expect(workflows).toHaveLength(5)
+    for (const workflow of workflows) {
+      expect(workflow).toStrictEqual({
         id: expect.stringMatching(
           /^payment-settler-0xMockTransactionHash-\d+$/,
         ),
@@ -412,11 +403,17 @@ describe('payment settler scheduled handler', () => {
           transactionHash: '0xMockTransactionHash',
           metadata: {
             onSuccess: 'settlement-confirmed',
-            successData: { dataSetIds: [id3] },
-            retryData: { dataSetIds: [id3] },
+            successData: {
+              settlementType: expect.stringMatching(/^(cache-miss|bandwidth)$/),
+              ids: [expect.any(String)],
+            },
+            retryData: {
+              settlementType: expect.stringMatching(/^(cache-miss|bandwidth)$/),
+              ids: [expect.any(String)],
+            },
           },
         },
-      },
-    ])
+      })
+    }
   })
 })
diff --git a/usage-reporter/bin/usage-reporter.js b/usage-reporter/bin/usage-reporter.js
index fc80a910..2f9249b7 100644
--- a/usage-reporter/bin/usage-reporter.js
+++ b/usage-reporter/bin/usage-reporter.js
@@ -56,15 +56,18 @@ export default {
       )
       console.log(`Aggregating usage data up to timestamp: ${upToTimestampMs}`)
 
-      // Aggregate usage data for all datasets that need reporting
+      // Aggregate usage data per data set. The contract resolves each data set
+      // to its shared cdn_rail_id and aggregates bandwidth onto that rail.
       const usageData = await aggregateUsageData(env.DB, upToTimestampMs)
 
-      if (usageData.length === 0) {
+      if (usageData.dataSetIds.length === 0) {
         console.log('No usage data found')
         return
       }
 
-      console.log(`Found usage data for ${usageData.length} data sets`)
+      console.log(
+        `Found usage data for ${usageData.dataSetIds.length} data sets`,
+      )
 
       // Prepare usage report data for contract call
       const usageReportData = prepareUsageReportData(usageData)
@@ -90,7 +93,7 @@ export default {
       })
 
       console.log(
-        `Sending recordUsageRollups transaction for ${usageReportData.dataSetIds.length} data sets`,
+        `Sending recordUsageRollups transaction for ${usageData.dataSetIds.length} data sets`,
       )
 
       // Send transaction
@@ -98,9 +101,9 @@ export default {
 
       console.log(`Transaction sent: ${hash}`)
 
-      // Store transaction hash to prevent double-counting
+      // Store transaction hash for every contributing data set to prevent double-counting
       await env.DB.batch(
-        usageReportData.dataSetIds.map((dataSetId) =>
+        usageData.dataSetIds.map((dataSetId) =>
           env.DB.prepare(
             `UPDATE data_sets SET pending_usage_report_tx_hash = ? WHERE id = ?`,
           ).bind(hash, dataSetId),
@@ -108,7 +111,7 @@ export default {
       )
 
       console.log(
-        `Stored pending transaction hash for ${usageReportData.dataSetIds.length} data sets`,
+        `Stored pending transaction hash for ${usageData.dataSetIds.length} data sets`,
       )
 
       const upToTimestamp = new Date(upToTimestampMs).toISOString()
diff --git a/usage-reporter/lib/usage-report.js b/usage-reporter/lib/usage-report.js
index 9a018f9c..1b457b52 100644
--- a/usage-reporter/lib/usage-report.js
+++ b/usage-reporter/lib/usage-report.js
@@ -1,6 +1,13 @@
 /**
- * Aggregate usage data, for all data sets, between last reported timestamp and
- * a target timestamp
+ * Aggregate usage data per data set, between each data set's last reported
+ * timestamp and a target timestamp.
+ *
+ * Usage is reported per data set: `cdn_bytes` is all egress (cache hits and
+ * misses), `cache_miss_bytes` is the cache-miss subset. The FilBeamOperator
+ * contract resolves each data set to its shared `cdn_rail_id` and aggregates
+ * the bandwidth onto that rail, so multiple data sets in one CDN subscription
+ * settle bandwidth once. Cache-miss stays per data set (each copy is served by
+ * a different provider, a different payee).
  *
  * @param {D1Database} db
  * @param {number} upToTimestampMs - Target timestamp in milliseconds
@@ -12,16 +19,12 @@
  *   }[]
  * >}
  */
-export async function aggregateUsageData(db, upToTimestampMs) {
-  // Query aggregates total usage data between usage_reported_until and upToTimestampMs
-  // Returns sum of CDN bytes, cache-miss bytes, and max timestamp for each dataset
-  // Excludes datasets with pending transactions to prevent double-counting
+export async function aggregateUsageByDataSet(db, upToTimestampMs) {
   const upToTimestampIso = new Date(upToTimestampMs).toISOString()
   const query = `
     SELECT
       rl.data_set_id,
-      -- Note: cdn_bytes tracks all egress (cache hits + cache misses)
-      -- cache_miss_bytes tracks only cache misses (subset of cdn_bytes)
+      -- cdn_bytes tracks all egress (cache hits + cache misses)
       SUM(rl.egress_bytes) as cdn_bytes,
       SUM(CASE WHEN rl.cache_miss = 1 AND rl.cache_miss_response_valid = 1 THEN rl.egress_bytes ELSE 0 END) as cache_miss_bytes
     FROM retrieval_logs rl
@@ -32,7 +35,7 @@ export async function aggregateUsageData(db, upToTimestampMs) {
       AND rl.bot_name IS NULL
       AND ds.pending_usage_report_tx_hash IS NULL
     GROUP BY rl.data_set_id
-    HAVING (cdn_bytes > 0 OR cache_miss_bytes > 0)
+    HAVING cdn_bytes > 0
   `
 
   const results = /**
@@ -51,25 +54,56 @@ export async function aggregateUsageData(db, upToTimestampMs) {
 }
 
 /**
- * Prepare usage report data for FilBeam contract call
+ * Aggregate usage data, for all data sets, between last reported timestamp and
+ * a target timestamp.
+ *
+ * Usage is reported per data set. dataSetIds lists every data set that
+ * contributed, used to advance the per-data-set usage_reported_until
+ * watermark.
+ *
+ * @param {D1Database} db
+ * @param {number} upToTimestampMs - Target timestamp in milliseconds
+ * @returns {Promise<{
+ *   usageByDataSet: {
+ *     data_set_id: string
+ *     cdn_bytes: number
+ *     cache_miss_bytes: number
+ *   }[]
+ *   dataSetIds: string[]
+ * }>}
+ */
+export async function aggregateUsageData(db, upToTimestampMs) {
+  const usageByDataSet = await aggregateUsageByDataSet(db, upToTimestampMs)
+  const dataSetIds = usageByDataSet.map((usage) => String(usage.data_set_id))
+
+  return { usageByDataSet, dataSetIds }
+}
+
+/**
+ * Prepare usage report data for the FilBeam contract call.
+ *
+ * Produces three parallel arrays aligned by data set, matching the contract's
+ * `recordUsageRollups(toEpoch, dataSetIds, cdnBytesUsed, cacheMissBytesUsed)`.
  *
  * @param {{
- *   data_set_id: string
- *   cdn_bytes: number
- *   cache_miss_bytes: number
- * }[]} usageData
+ *   usageByDataSet: {
+ *     data_set_id: string
+ *     cdn_bytes: number
+ *     cache_miss_bytes: number
+ *   }[]
+ * }} usageData
  * @returns {{
  *   dataSetIds: string[]
  *   cdnBytesUsed: bigint[]
  *   cacheMissBytesUsed: bigint[]
  * }}
  */
-export function prepareUsageReportData(usageData) {
+export function prepareUsageReportData({ usageByDataSet }) {
   const dataSetIds = []
   const cdnBytesUsed = []
   const cacheMissBytesUsed = []
 
-  for (const usage of usageData) {
+  for (const usage of usageByDataSet) {
     dataSetIds.push(usage.data_set_id)
     cdnBytesUsed.push(BigInt(usage.cdn_bytes))
     cacheMissBytesUsed.push(BigInt(usage.cache_miss_bytes))
diff --git a/usage-reporter/test/test-helpers.js b/usage-reporter/test/test-helpers.js
index cfc0078e..7efb72e2 100644
--- a/usage-reporter/test/test-helpers.js
+++ b/usage-reporter/test/test-helpers.js
@@ -62,6 +62,8 @@ export async function withDataSet(
     terminateServiceTxHash = null,
     usageReportedUntil = '1970-01-01T00:00:00.000Z',
     pendingUsageReportTxHash = null,
+    cdnRailId = `rail-${id}`,
+    cacheMissRailId = `cache-miss-rail-${id}`,
   },
 ) {
   // Ensure service provider exists
@@ -72,7 +74,7 @@ export async function withDataSet(
     .run()
 
   await env.DB.prepare(
-    `INSERT INTO data_sets (id, service_provider_id, payer_address, with_cdn, terminate_service_tx_hash, usage_reported_until, pending_usage_report_tx_hash) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+    `INSERT INTO data_sets (id, service_provider_id, payer_address, with_cdn, terminate_service_tx_hash, usage_reported_until, pending_usage_report_tx_hash, cdn_rail_id, cache_miss_rail_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
   )
     .bind(
       String(id),
@@ -82,6 +84,8 @@ export async function withDataSet(
       terminateServiceTxHash,
       usageReportedUntil,
       pendingUsageReportTxHash,
+      cdnRailId,
+      cacheMissRailId,
     )
     .run()
 }
diff --git a/usage-reporter/test/usage-report.test.js b/usage-reporter/test/usage-report.test.js
index 47a7973f..601c7113 100644
--- a/usage-reporter/test/usage-report.test.js
+++ b/usage-reporter/test/usage-report.test.js
@@ -2,6 +2,7 @@ import { describe, it, expect, afterEach } from 'vitest'
 import { env } from 'cloudflare:test'
 import {
   aggregateUsageData,
+  aggregateUsageByDataSet,
   prepareUsageReportData,
 } from '../lib/usage-report.js'
 import {
@@ -22,16 +23,19 @@ describe('usage report', () => {
     })
 
     describe('aggregateUsageData', () => {
-      it('should aggregate usage data by cache miss status', async () => {
+      it('aggregates cdn and cache-miss bytes per data set', async () => {
         await withDataSet(env, {
           id: '1',
+          cdnRailId: 'rail-1',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
         })
         await withDataSet(env, {
           id: '2',
+          cdnRailId: 'rail-2',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
         })
 
+        // Excluded: timestamp equals usage_reported_until (not strictly greater)
         await withRetrievalLog(env, {
           timestamp: EPOCH_99_TIMESTAMP_ISO,
           dataSetId: '1',
@@ -62,6 +66,7 @@ describe('usage report', () => {
           cacheMissResponseValid: 1,
         })
 
+        // Excluded: timestamp after the target
         await withRetrievalLog(env, {
           timestamp: EPOCH_101_TIMESTAMP_ISO,
           dataSetId: '1',
@@ -69,35 +74,68 @@ describe('usage report', () => {
           cacheMiss: 0,
         })
 
+        const usageData = await aggregateUsageData(
+          env.DB,
+          EPOCH_100_TIMESTAMP_MS,
+        )
+
+        expect(usageData).toStrictEqual({
+          usageByDataSet: [
+            { data_set_id: '1', cdn_bytes: 2500, cache_miss_bytes: 500 },
+            { data_set_id: '2', cdn_bytes: 3000, cache_miss_bytes: 3000 },
+          ],
+          dataSetIds: ['1', '2'],
+        })
+      })
+
+      it('reports each member of a shared CDN group as its own data set', async () => {
+        await withDataSet(env, {
+          id: '1',
+          cdnRailId: 'shared-rail',
+          usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
+        })
+        await withDataSet(env, {
+          id: '2',
+          cdnRailId: 'shared-rail',
+          usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
+        })
+
         await withRetrievalLog(env, {
-          timestamp: EPOCH_101_TIMESTAMP_ISO,
-          dataSetId: '2',
-          egressBytes: 9999,
+          timestamp: EPOCH_100_TIMESTAMP_ISO,
+          dataSetId: '1',
+          egressBytes: 1000,
           cacheMiss: 0,
         })
+        await withRetrievalLog(env, {
+          timestamp: EPOCH_100_TIMESTAMP_ISO,
+          dataSetId: '2',
+          egressBytes: 2000,
+          cacheMiss: 1,
+          cacheMissResponseValid: 1,
+        })
 
         const usageData = await aggregateUsageData(
           env.DB,
           EPOCH_100_TIMESTAMP_MS,
         )
 
-        expect(usageData).toStrictEqual([
-          {
-            data_set_id: '1',
-            cdn_bytes: 2500,
-            cache_miss_bytes: 500,
-          },
-          {
-            data_set_id: '2',
-            cdn_bytes: 3000,
-            cache_miss_bytes: 3000,
-          },
-        ])
+        // Bandwidth is aggregated onto the shared rail on-chain; the worker
+        // reports per data set so each contributes its own egress.
+        expect(usageData).toStrictEqual({
+          usageByDataSet: [
+            { data_set_id: '1', cdn_bytes: 1000, cache_miss_bytes: 0 },
+            { data_set_id: '2', cdn_bytes: 2000, cache_miss_bytes: 2000 },
+          ],
+          dataSetIds: ['1', '2'],
+        })
       })
+    })
 
+    describe('aggregateUsageByDataSet', () => {
       it('should include non-200 responses but filter out null egress_bytes', async () => {
         await withDataSet(env, {
           id: '1',
+          cdnRailId: 'rail-1',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
         })
 
@@ -134,32 +172,31 @@ describe('usage report', () => {
           cacheMissResponseValid: 1,
         })
 
-        const usageData = await aggregateUsageData(
+        const usage = await aggregateUsageByDataSet(
           env.DB,
           EPOCH_100_TIMESTAMP_MS,
         )
 
-        expect(usageData).toStrictEqual([
-          {
-            data_set_id: '1',
-            cdn_bytes: 1800,
-            cache_miss_bytes: 300,
-          },
+        expect(usage).toStrictEqual([
+          { data_set_id: '1', cdn_bytes: 1800, cache_miss_bytes: 300 },
         ])
       })
 
       it('should only aggregate data for datasets with usage_reported_until < upToTimestamp', async () => {
-        await withDataSet(env, { id: '1' })
+        await withDataSet(env, { id: '1', cdnRailId: 'rail-1' })
         await withDataSet(env, {
           id: '2',
+          cdnRailId: 'rail-2',
           usageReportedUntil: EPOCH_98_TIMESTAMP_ISO,
         })
         await withDataSet(env, {
           id: '3',
+          cdnRailId: 'rail-3',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
         })
         await withDataSet(env, {
           id: '4',
+          cdnRailId: 'rail-4',
           usageReportedUntil: EPOCH_100_TIMESTAMP_ISO,
         })
 
@@ -172,94 +209,29 @@ describe('usage report', () => {
           })
         }
 
-        const usageData = await aggregateUsageData(
-          env.DB,
-          EPOCH_100_TIMESTAMP_MS,
-        )
-
-        expect(usageData).toStrictEqual([
-          {
-            data_set_id: '1',
-            cdn_bytes: 1000,
-            cache_miss_bytes: 0,
-          },
-          {
-            data_set_id: '2',
-            cdn_bytes: 1000,
-            cache_miss_bytes: 0,
-          },
-          {
-            data_set_id: '3',
-            cdn_bytes: 1000,
-            cache_miss_bytes: 0,
-          },
-        ])
-      })
-
-      it('should filter out datasets with zero cdn and cache-miss bytes', async () => {
-        await withDataSet(env, {
-          id: '1',
-          usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
-        })
-        await withDataSet(env, {
-          id: '2',
-          usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
-        })
-        await withDataSet(env, {
-          id: '3',
-          usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
-        })
-
-        await withRetrievalLog(env, {
-          timestamp: EPOCH_100_TIMESTAMP_ISO,
-          dataSetId: '1',
-          egressBytes: 1000,
-          cacheMiss: 0,
-        })
-
-        await withRetrievalLog(env, {
-          timestamp: EPOCH_100_TIMESTAMP_ISO,
-          dataSetId: '2',
-          egressBytes: null,
-          cacheMiss: 0,
-        })
-
-        await withRetrievalLog(env, {
-          timestamp: EPOCH_100_TIMESTAMP_ISO,
-          dataSetId: '3',
-          egressBytes: 500,
-          cacheMiss: 1,
-          cacheMissResponseValid: 1,
-        })
-
-        const usageData = await aggregateUsageData(
+        const usage = await aggregateUsageByDataSet(
           env.DB,
           EPOCH_100_TIMESTAMP_MS,
         )
 
-        expect(usageData).toStrictEqual([
-          {
-            data_set_id: '1',
-            cdn_bytes: 1000,
-            cache_miss_bytes: 0,
-          },
-          {
-            data_set_id: '3',
-            cdn_bytes: 500,
-            cache_miss_bytes: 500,
-          },
+        expect(usage).toStrictEqual([
+          { data_set_id: '1', cdn_bytes: 1000, cache_miss_bytes: 0 },
+          { data_set_id: '2', cdn_bytes: 1000, cache_miss_bytes: 0 },
+          { data_set_id: '3', cdn_bytes: 1000, cache_miss_bytes: 0 },
         ])
       })
 
       it('should exclude datasets with pending usage report transactions', async () => {
         await withDataSet(env, {
           id: '1',
+          cdnRailId: 'rail-1',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
           pendingUsageReportTxHash: null,
         })
 
         await withDataSet(env, {
           id: '2',
+          cdnRailId: 'rail-2',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
           pendingUsageReportTxHash: '0x123abc',
         })
@@ -278,21 +250,17 @@ describe('usage report', () => {
           cacheMiss: 0,
         })
 
-        const usageData = await aggregateUsageData(
+        const usage = await aggregateUsageByDataSet(
           env.DB,
           EPOCH_100_TIMESTAMP_MS,
         )
 
-        expect(usageData).toStrictEqual([
-          {
-            data_set_id: '1',
-            cdn_bytes: 1000,
-            cache_miss_bytes: 0,
-          },
+        expect(usage).toStrictEqual([
+          { data_set_id: '1', cdn_bytes: 1000, cache_miss_bytes: 0 },
         ])
       })
 
-      it('should not count invalid cache miss responses towards cache miss bytes', async () => {
+      it('should only count valid cache miss responses', async () => {
         await withDataSet(env, {
           id: '1',
           usageReportedUntil: EPOCH_99_TIMESTAMP_ISO,
@@ -307,76 +275,35 @@ describe('usage report', () => {
           cacheMissResponseValid: 0,
         })
 
-        const usageData = await aggregateUsageData(
+        const usage = await aggregateUsageByDataSet(
           env.DB,
           EPOCH_100_TIMESTAMP_MS,
         )
 
-        expect(usageData).toStrictEqual([
-          {
-            data_set_id: '1',
-            cdn_bytes: 1000,
-            cache_miss_bytes: 0,
-          },
+        // Egress still counts toward bandwidth, but invalid cache-miss does not
+        expect(usage).toStrictEqual([
+          { data_set_id: '1', cdn_bytes: 1000, cache_miss_bytes: 0 },
         ])
       })
     })
   })
 
   describe('prepareUsageReportData', () => {
-    it('should prepare batch data for contract call', () => {
-      const usageData = [
-        {
-          data_set_id: '1',
-          cdn_bytes: 1000,
-          cache_miss_bytes: 500,
-        },
-        {
-          data_set_id: '2',
-          cdn_bytes: 2000,
-          cache_miss_bytes: 0,
-        },
-        {
-          data_set_id: '3',
-          cdn_bytes: 0,
-          cache_miss_bytes: 3000,
-        },
-      ]
-
-      const batchData = prepareUsageReportData(usageData)
-
-      expect(batchData).toStrictEqual({
-        dataSetIds: ['1', '2', '3'],
-        cdnBytesUsed: [1000n, 2000n, 0n],
-        cacheMissBytesUsed: [500n, 0n, 3000n],
-      })
-    })
-
-    it('should process all datasets', () => {
-      const usageData = [
-        {
-          data_set_id: '1',
-          cdn_bytes: 1000,
-          cache_miss_bytes: 500,
-        },
-        {
-          data_set_id: '2',
-          cdn_bytes: 2000,
-          cache_miss_bytes: 0,
-        },
-        {
-          data_set_id: '3',
-          cdn_bytes: 0,
-          cache_miss_bytes: 3000,
-        },
-      ]
+    it('produces parallel per-data-set arrays for recordUsageRollups', () => {
+      const usageData = {
+        usageByDataSet: [
+          { data_set_id: '1', cdn_bytes: 1000, cache_miss_bytes: 500 },
+          { data_set_id: '3', cdn_bytes: 2000, cache_miss_bytes: 3000 },
+        ],
+        dataSetIds: ['1', '3'],
+      }
 
       const batchData = prepareUsageReportData(usageData)
 
       expect(batchData).toStrictEqual({
-        dataSetIds: ['1', '2', '3'],
-        cdnBytesUsed: [1000n, 2000n, 0n],
-        cacheMissBytesUsed: [500n, 0n, 3000n],
+        dataSetIds: ['1', '3'],
+        cdnBytesUsed: [1000n, 2000n],
+        cacheMissBytesUsed: [500n, 3000n],
       })
     })
   })