processNewValue handles local updates (TryUpdate, applyRefreshResult) where we don't have an external timestamp. The ignoreRollbackOf parameter controls rollback behavior: - ignoreRollbackOf = 0: Accept all changes (used for earliest block - can decrease due to pruning) - ignoreRollbackOf > 0: Ignor
(source string, newVal int64)
| 270 | // |
| 271 | // This is different from processNewState which uses timestamp ordering for remote reconciliation. |
| 272 | func (c *counterInt64) processNewValue(source string, newVal int64) bool { |
| 273 | currentValue := c.value.Load() |
| 274 | if newVal == currentValue { |
| 275 | // Value is the same, but mark as fresh to prevent repeated refresh attempts |
| 276 | c.allocateUpdatedAtMs() |
| 277 | return false |
| 278 | } |
| 279 | |
| 280 | if newVal > currentValue { |
| 281 | // Forward progress: always accept |
| 282 | for { |
| 283 | if c.value.CompareAndSwap(currentValue, newVal) { |
| 284 | c.allocateUpdatedAtMs() // Mark as fresh |
| 285 | c.registry.logger.Trace(). |
| 286 | Str("source", source). |
| 287 | Str("key", c.key). |
| 288 | Int64("from", currentValue). |
| 289 | Int64("to", newVal). |
| 290 | Msg("counter value increased (local)") |
| 291 | c.triggerValueCallback(newVal) |
| 292 | return true |
| 293 | } |
| 294 | currentValue = c.value.Load() |
| 295 | if newVal <= currentValue { |
| 296 | return false |
| 297 | } |
| 298 | } |
| 299 | } |
| 300 | |
| 301 | // Rollback handling: only accept rollbacks where gap exceeds the threshold. |
| 302 | // - ignoreRollbackOf=0 (earliest): gap > 0 is always true, so ALL rollbacks accepted |
| 303 | // - ignoreRollbackOf=1024 (latest/finalized): only accept large rollbacks (real reorgs) |
| 304 | gap := currentValue - newVal |
| 305 | |
| 306 | if gap <= c.ignoreRollbackOf { |
| 307 | // Small rollback within threshold - ignore as noise |
| 308 | c.registry.logger.Trace(). |
| 309 | Str("source", source). |
| 310 | Str("key", c.key). |
| 311 | Int64("currentValue", currentValue). |
| 312 | Int64("newVal", newVal). |
| 313 | Int64("gap", gap). |
| 314 | Int64("ignoreRollbackOf", c.ignoreRollbackOf). |
| 315 | Msg("small rollback ignored (local)") |
| 316 | return false |
| 317 | } |
| 318 | |
| 319 | // Large rollback exceeds threshold - apply it and trigger callback |
| 320 | for { |
| 321 | if c.value.CompareAndSwap(currentValue, newVal) { |
| 322 | c.allocateUpdatedAtMs() // Mark as fresh |
| 323 | c.registry.logger.Trace(). |
| 324 | Str("source", source). |
| 325 | Str("key", c.key). |
| 326 | Int64("from", currentValue). |
| 327 | Int64("to", newVal). |
| 328 | Int64("gap", gap). |
| 329 | Msg("large rollback applied (local)") |