| 122 | } |
| 123 | |
| 124 | func (c *counterInt64) processNewState(source string, st CounterInt64State) bool { |
| 125 | // processNewState handles updates from remote sources (WatchCounterInt64, reconciliation). |
| 126 | // It applies the SAME rollback handling as processNewValue for consistency: |
| 127 | // - ignoreRollbackOf = 0: Accept all changes (for earliest block) |
| 128 | // - ignoreRollbackOf > 0: Ignore small rollbacks (for latest/finalized - avoids noise) |
| 129 | // |
| 130 | // This ensures consistent "only-increasing" semantics for latest/finalized across |
| 131 | // all instances, preventing noise from propagating via shared state. |
| 132 | |
| 133 | // Track if this is a remote update (with external timestamp) vs local update. |
| 134 | // For remote updates, we only CHECK timestamp ordering here - we DON'T update our local |
| 135 | // timestamp until we know we're going to accept the value change (forward progress or large rollback). |
| 136 | // |
| 137 | // SPECIAL CASE - Small rollback rejection: When we reject a small rollback from a remote update |
| 138 | // with a newer timestamp, we advance our local timestamp past the remote's timestamp. This ensures |
| 139 | // our higher local value will "win" reconciliation in scheduleBackgroundPushCurrent and get pushed |
| 140 | // to remote, since local.UpdatedAt will be > remote.UpdatedAt. |
| 141 | isRemoteUpdate := st.UpdatedAt > 0 |
| 142 | |
| 143 | if !isRemoteUpdate { |
| 144 | // For local updates, allow UpdatedAt=0 and allocate a unique timestamp. |
| 145 | st.UpdatedAt = c.allocateUpdatedAtMs() |
| 146 | } else { |
| 147 | // For remote updates, just check timestamp ordering - don't update local timestamp yet. |
| 148 | currentTs := c.updatedAtMs() |
| 149 | if st.UpdatedAt <= currentTs { |
| 150 | // Reject stale/out-of-order states by UpdatedAt ordering. |
| 151 | // This provides idempotency and prevents stale remote values from overriding fresh local detections. |
| 152 | return false |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | currentValue := c.value.Load() |
| 157 | newVal := st.Value |
| 158 | |
| 159 | if newVal == currentValue { |
| 160 | // Values appear equal - no value change needed. |
| 161 | // For remote updates, we want to update timestamp since remote is fresher. |
| 162 | // However, we must verify the value hasn't changed concurrently to avoid a TOCTOU race: |
| 163 | // a concurrent TryUpdate could change the value between our Load() and this check, |
| 164 | // causing us to incorrectly set the remote's stale timestamp for a different local value. |
| 165 | if isRemoteUpdate { |
| 166 | if c.updateUpdatedAtMs(st.UpdatedAt) { |
| 167 | // Timestamp was updated. Verify value is still what we expect. |
| 168 | // If a concurrent update changed the value after our initial Load(), |
| 169 | // we need to ensure our timestamp reflects that change, not the remote's stale state. |
| 170 | if c.value.Load() != newVal { |
| 171 | // Value changed concurrently - allocate fresh timestamp to ensure |
| 172 | // local state wins in reconciliation. This maintains the invariant that |
| 173 | // timestamps accurately reflect when values were last updated locally. |
| 174 | c.allocateUpdatedAtMs() |
| 175 | } |
| 176 | } |
| 177 | } |
| 178 | return false |
| 179 | } |
| 180 | |
| 181 | if newVal > currentValue { |