refactor(wotoken-yield,es-address-yield): deterministic day-end sweeps

Both processors swept passive holders on ctx.latestBlockOfDay, which keys off per-batch
lastBlockPerDay/blocks.at(-1) and therefore fires on batch boundaries — non-deterministic
timing and, during backfill, one previewRedeem/accRewardPerShare + a full holder-rewrite
per batch instead of per day (the RPC/write bloat seen live).

Add forEachBlockByDay: a driver that closes out each UTC day once, at that day's true
last block (next block is a different day, with a cross-batch carry so a boundary on a
batch split still lands on the real last block), and only sweeps the batch tail when
ctx.isHead — keeping the current day fresh live without redundant backfill sweeps.

Validated on the xOGN launch window (mainnet 19919745->19990000): a single run and a run
split at 19955000 produce byte-identical completed-day data (checksum f3e9b04d…), each
completed day's active holders share one day-end block, zero negatives.

Author: apexearth

src/templates/exponential-staking/es-address-yield.ts

@@ -1,6 +1,7 @@
 import * as abi from '@abi/exponential-staking'
 import { ESAddressYield } from '@model'
 import { Block, Context, EvmBatchProcessor, Processor, logFilter } from '@originprotocol/squid-utils'
+import { DayBoundaryCarry, forEachBlockByDay } from '@utils/for-each-block-by-day'
 
 // xOGN is a MasterChef-style accumulator: previewRewards(user) == balanceOf(user) *
 // (accRewardPerShare - rewardDebtPerShare(user)) / 1e12. So accRewardPerShare is scaled by 1e12 and a
@@ -39,6 +40,7 @@ export const createESAddressYieldProcessor = ({
   // Persisted across batches, seeded once from persisted rows (see `hydrate`).
   const currentDayRows = new Map<string, ESAddressYield>()
   const previousDayRows = new Map<string, ESAddressYield>()
+  const dayCarry: DayBoundaryCarry = {}
   let hydrated = false
 
   const hydrate = async (ctx: Context) => {
@@ -67,8 +69,9 @@ export const createESAddressYieldProcessor = ({
     name: `xOGN Address Yield ${staking}`,
     from,
     setup: (p: EvmBatchProcessor) => {
-      // includeAllBlocks so the daily forced checkpoint can fire on the last block of each day even
-      // when no stake/unstake occurred (matches the ARM/otoken batch).
+      // includeAllBlocks so the day-boundary detection sees consecutive blocks (and thus every true
+      // end-of-day) even on days with no stake/unstake. Redundant on mainnet (already all-blocks from
+      // block 15M) but keeps this processor self-sufficient.
       p.includeAllBlocks({ from })
       p.addLog(stakeFilter.value)
       p.addLog(unstakeFilter.value)
@@ -143,28 +146,30 @@ export const createESAddressYieldProcessor = ({
         changedYieldRows.set(row.id, row)
       }
 
-      for (const block of ctx.blocks) {
-        for (const log of block.logs) {
-          if (stakeFilter.matches(log)) {
-            // Stake mints `points` xOGN and locks `amount` OGN (new stake or the relock half of an
-            // extend). Accrue at the old balance first, then apply both deltas.
-            const event = abi.events.Stake.decode(log)
-            const R = await getR(block)
-            checkpoint(event.user, block, R, event.points, event.amount)
-          } else if (unstakeFilter.matches(log)) {
-            // Unstake burns `points` xOGN and withdraws OGN principal. `amount` is the post-penalty
-            // withdrawn amount; the full-exit clamp above zeroes any residual once points hit 0.
-            const event = abi.events.Unstake.decode(log)
-            const R = await getR(block)
-            checkpoint(event.user, block, R, -event.points, -event.amount)
+      // Deterministic per-day accrual: stake/unstake checkpoint at exact event time; each day's pure
+      // reward accrual is closed out once, at that day's true last block (`onDayEnd`), regardless of
+      // how blocks are batched. At the chain head the current day is also swept for freshness.
+      await forEachBlockByDay(ctx, dayCarry, {
+        onBlock: async (block) => {
+          for (const log of block.logs) {
+            if (stakeFilter.matches(log)) {
+              // Stake mints `points` xOGN and locks `amount` OGN (new stake or the relock half of an
+              // extend). Accrue at the old balance first, then apply both deltas.
+              const event = abi.events.Stake.decode(log)
+              const R = await getR(block)
+              checkpoint(event.user, block, R, event.points, event.amount)
+            } else if (unstakeFilter.matches(log)) {
+              // Unstake burns `points` xOGN and withdraws OGN principal. `amount` is the post-penalty
+              // withdrawn amount; the full-exit clamp above zeroes any residual once points hit 0.
+              const event = abi.events.Unstake.decode(log)
+              const R = await getR(block)
+              checkpoint(event.user, block, R, -event.points, -event.amount)
+            }
           }
-        }
-
-        // Daily forced checkpoint: close out each day's accrual for every staker with a live balance,
-        // including passive ones. Skip when there are no active stakers so we never call the contract
-        // before it has any (keeps `from` robust). `latestBlockOfDay` fires on each day's last block
-        // and the batch tail, so one sweep/day completes the series and the tail keeps today fresh.
-        if (block.header.height >= from && ctx.latestBlockOfDay(block)) {
+        },
+        onDayEnd: async (block) => {
+          if (block.header.height < from) return
+          // Skip when there are no active stakers so we never call the contract before it has any.
           const activeStakers: string[] = []
           for (const account of new Set([...currentDayRows.keys(), ...previousDayRows.keys()])) {
             const seedRow = currentDayRows.get(account) ?? previousDayRows.get(account)
@@ -174,8 +179,8 @@ export const createESAddressYieldProcessor = ({
             const R = await getR(block)
             for (const account of activeStakers) checkpoint(account, block, R, 0n, 0n)
           }
-        }
-      }
+        },
+      })
 
       await ctx.store.upsert([...changedYieldRows.values()])
     },

src/templates/wotoken-yield/wotoken-yield.ts

@@ -2,6 +2,7 @@ import * as wotokenAbi from '@abi/woeth'
 import { WOTokenAddressYield } from '@model'
 import { Block, Context, EvmBatchProcessor, Processor, logFilter } from '@originprotocol/squid-utils'
 import { ADDRESS_ZERO } from '@utils/addresses'
+import { DayBoundaryCarry, forEachBlockByDay } from '@utils/for-each-block-by-day'
 
 const ONE = 10n ** 18n
 
@@ -47,6 +48,7 @@ export const createWOTokenYieldProcessor = ({
   // Persisted across batches, seeded once from persisted rows (see `hydrate`).
   const currentDayRows = new Map<string, WOTokenAddressYield>()
   const previousDayRows = new Map<string, WOTokenAddressYield>()
+  const dayCarry: DayBoundaryCarry = {}
   let hydrated = false
   let warnedNegativeBalance = false
 
@@ -77,8 +79,9 @@ export const createWOTokenYieldProcessor = ({
     name,
     from,
     setup: (p: EvmBatchProcessor) => {
-      // includeAllBlocks so the daily forced checkpoint can fire on frequency/end-of-day blocks
-      // even when no wrapped transfer occurred that day (matches the ARM/otoken-state batch).
+      // includeAllBlocks so the day-boundary detection sees consecutive blocks (and thus every true
+      // end-of-day) even on days with no wrapped transfer. Redundant on these chains — the OToken
+      // processor already includes all blocks from an earlier block — but keeps this self-sufficient.
       p.includeAllBlocks({ from })
       p.addLog(transferFilter.value)
     },
@@ -171,33 +174,32 @@ export const createWOTokenYieldProcessor = ({
         changedYieldRows.set(row.id, row)
       }
 
-      for (const block of ctx.blocks) {
-        for (const log of block.logs) {
-          if (transferFilter.matches(log)) {
-            // Mint = from 0 (wrap); burn = to 0 (unwrap); peer transfer otherwise.
-            // Every inbound (wrap or peer-in) marks cost basis at current R: for a wrap you deposit
-            // `shares * R` underlying assets; a peer-in is acquired at current market value. Outbounds
-            // reduce cost basis proportionally (handled inside checkpoint).
-            const event = wotokenAbi.events.Transfer.decode(log)
-            const R = await getR(block)
-            const fromZero = event.from.toLowerCase() === ADDRESS_ZERO
-            const toZero = event.to.toLowerCase() === ADDRESS_ZERO
-            if (!fromZero) {
-              checkpoint(event.from, block, R, -event.value)
-            }
-            if (!toZero) {
-              checkpoint(event.to, block, R, event.value, (event.value * R) / ONE)
+      // Deterministic per-day accrual: transfers checkpoint at exact event time; each day's pure
+      // share-appreciation is closed out once, at that day's true last block (`onDayEnd`), regardless
+      // of how blocks are batched. At the chain head the current day is also swept for freshness.
+      await forEachBlockByDay(ctx, dayCarry, {
+        onBlock: async (block) => {
+          for (const log of block.logs) {
+            if (transferFilter.matches(log)) {
+              // Mint = from 0 (wrap); burn = to 0 (unwrap); peer transfer otherwise.
+              // Every inbound (wrap or peer-in) marks cost basis at current R: for a wrap you deposit
+              // `shares * R` underlying assets; a peer-in is acquired at current market value.
+              // Outbounds reduce cost basis proportionally (handled inside checkpoint).
+              const event = wotokenAbi.events.Transfer.decode(log)
+              const R = await getR(block)
+              const fromZero = event.from.toLowerCase() === ADDRESS_ZERO
+              const toZero = event.to.toLowerCase() === ADDRESS_ZERO
+              if (!fromZero) {
+                checkpoint(event.from, block, R, -event.value)
+              }
+              if (!toZero) {
+                checkpoint(event.to, block, R, event.value, (event.value * R) / ONE)
+              }
             }
           }
-        }
-
-        // Daily forced checkpoint: close out each day's pure share-appreciation accrual for every
-        // known holder, including passive ones with no events. `latestBlockOfDay` is true on the last
-        // block of each day *and* the last block of the batch, so one sweep per day completes the
-        // daily series during backfill and the batch-tail sweep keeps the current day fresh while
-        // live. Intra-day sweeps would only overwrite the same dated row, so they are unnecessary
-        // here (unlike ARM, which also needs sub-daily ArmState snapshots).
-        if (block.header.height >= from && ctx.latestBlockOfDay(block)) {
+        },
+        onDayEnd: async (block) => {
+          if (block.header.height < from) return
           // Collect holders with a live balance first, and skip the block entirely when there are
           // none. Before the vault's first deposit there are no holders and its ERC4626 views revert,
           // so we must not call previewRedeem (getR) yet. This keeps `from` robust: it can sit at (or
@@ -211,8 +213,8 @@ export const createWOTokenYieldProcessor = ({
             const R = await getR(block)
             for (const account of activeHolders) checkpoint(account, block, R, 0n)
           }
-        }
-      }
+        },
+      })
 
       await ctx.store.upsert([...changedYieldRows.values()])
     },

src/utils/for-each-block-by-day.ts

@@ -0,0 +1,47 @@
+import { Block, Context } from '@originprotocol/squid-utils'
+
+const utcDate = (block: Block) => new Date(block.header.timestamp).toISOString().slice(0, 10)
+
+export interface DayBoundaryCarry {
+  lastBlock?: Block
+}
+
+/**
+ * Deterministically drive per-day work over a batch, independent of batch boundaries.
+ *
+ * `onBlock` runs for every block in order. `onDayEnd(block)` runs once per UTC day, at that day's
+ * true last block — detected by the next block being a different UTC day. `carry` bridges a day
+ * boundary that lands exactly on a batch split, so the day is still finalized at its real last block
+ * (not wherever the batch happened to end). When the batch is at the chain head, the final (still
+ * in-progress) day is also swept at the latest block so the current day stays fresh while live.
+ *
+ * This replaces `ctx.latestBlockOfDay`, whose `lastBlockPerDay` / `blocks.at(-1)` are computed per
+ * batch and therefore fire on batch boundaries — non-deterministic timing and redundant work.
+ *
+ * `carry` must persist across batches: declare it once at processor scope.
+ */
+export const forEachBlockByDay = async (
+  ctx: Context,
+  carry: DayBoundaryCarry,
+  handlers: {
+    onBlock: (block: Block) => Promise<void>
+    onDayEnd: (dayEndBlock: Block) => Promise<void>
+  },
+) => {
+  const blocks = ctx.blocks
+  for (let i = 0; i < blocks.length; i++) {
+    const block = blocks[i]
+    const prev = i === 0 ? carry.lastBlock : blocks[i - 1]
+    // A new UTC day has started, so `prev` was the previous day's last block — finalize it there.
+    if (prev && utcDate(prev) !== utcDate(block)) {
+      await handlers.onDayEnd(prev)
+    }
+    await handlers.onBlock(block)
+  }
+  if (blocks.length > 0) {
+    const tail = blocks[blocks.length - 1]
+    carry.lastBlock = tail
+    // At the chain head the current day is incomplete; sweep the latest block for live freshness.
+    if (ctx.isHead) await handlers.onDayEnd(tail)
+  }
+}