Advance local state after observing a remote [`Hlc`]. Implements the standard HLC merge rule so that subsequent calls to [`next`](HlcGenerator::next) return timestamps strictly greater than any observed remote timestamp.
(&self, remote: Hlc)
| 200 | /// [`next`](HlcGenerator::next) return timestamps strictly greater than |
| 201 | /// any observed remote timestamp. |
| 202 | pub fn observe(&self, remote: Hlc) -> ArrayResult<()> { |
| 203 | let now_ms = Self::now_ms()?.min(MAX_PHYSICAL_MS); |
| 204 | |
| 205 | let mut guard = self.state.lock().map_err(|_| ArrayError::HlcLockPoisoned)?; |
| 206 | let (last_physical, last_logical) = *guard; |
| 207 | |
| 208 | let new_physical = now_ms.max(last_physical).max(remote.physical_ms); |
| 209 | let new_logical = if new_physical == last_physical && new_physical == remote.physical_ms { |
| 210 | // All three agree on physical; advance logical past both. |
| 211 | last_logical |
| 212 | .max(remote.logical) |
| 213 | .checked_add(1) |
| 214 | .ok_or_else(|| ArrayError::InvalidHlc { |
| 215 | detail: "logical counter overflow during observe".into(), |
| 216 | })? |
| 217 | } else if new_physical == last_physical { |
| 218 | last_logical |
| 219 | .checked_add(1) |
| 220 | .ok_or_else(|| ArrayError::InvalidHlc { |
| 221 | detail: "logical counter overflow during observe (local wins)".into(), |
| 222 | })? |
| 223 | } else if new_physical == remote.physical_ms { |
| 224 | remote |
| 225 | .logical |
| 226 | .checked_add(1) |
| 227 | .ok_or_else(|| ArrayError::InvalidHlc { |
| 228 | detail: "logical counter overflow during observe (remote wins)".into(), |
| 229 | })? |
| 230 | } else { |
| 231 | // now_ms is strictly larger than both; reset logical. |
| 232 | 0 |
| 233 | }; |
| 234 | |
| 235 | *guard = (new_physical, new_logical); |
| 236 | Ok(()) |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | #[cfg(test)] |