1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
//! Dispatch-without-emit gate — used by the server-as-replica path
//! to apply frames pulled from an upstream primary
//! without immediately re-pushing them into this shard's own
//! `ReplicationSource`. Without the gate, a server with both an
//! upstream link AND its own primary listener (chain replication, or
//! the brief overlap during `REPLICAOF NO ONE` promotion) would emit
//! every applied frame to its own downstream replicas, double-counting
//! the offset and creating infinite chains.
//!
//! Chain replication is explicitly out of scope,
//! so the gate is defensive: it documents intent
//! and prevents the misconfig — primary + REPLICAOF together — from
//! silently corrupting the downstream offset stream.
//!
//! Usage:
//!
//! ```ignore
//! let _g = kevy_rt::ReplicatedApplyGuard::enter();
//! // dispatch frame here — any post_write_housekeeping that hits
//! // this shard's ReplicationSource is suppressed for the duration
//! // of `_g`.
//! ```
//!
//! Scope: the gate suppresses ONLY the `ReplicationSource::push_mutation`
//! call inside [`crate::shard::Shard::post_write_housekeeping`]. AOF
//! append, WATCH version bump, keyspace notifications, and BLOCK wakes
//! all still fire — the local store state must remain correct for
//! anyone reading from this server.
use std::cell::Cell;
thread_local! {
/// `true` while a replicated-apply scope is active on this thread.
/// Set by [`ReplicatedApplyGuard::enter`], cleared on drop.
static APPLYING_REPLICATED: Cell<bool> = const { Cell::new(false) };
}
/// RAII guard that marks the current thread as "applying a replicated
/// frame" for the guard's lifetime. The replica runner
/// enters this scope before each `dispatch` call so the apply doesn't
/// re-push the frame into this shard's own backlog.
#[derive(Debug)]
pub struct ReplicatedApplyGuard {
/// Prior gate value — supports nesting (caller can enter a second
/// scope without losing the outer one's intent; drop restores).
prev: bool,
}
impl ReplicatedApplyGuard {
/// Enter a replicated-apply scope on the current thread. Nestable
/// — the inner guard restores the outer state on drop.
#[must_use = "ReplicatedApplyGuard is RAII — drop it at scope end"]
pub fn enter() -> Self {
let prev = APPLYING_REPLICATED.with(Cell::get);
APPLYING_REPLICATED.with(|c| c.set(true));
Self { prev }
}
}
impl Drop for ReplicatedApplyGuard {
fn drop(&mut self) {
APPLYING_REPLICATED.with(|c| c.set(self.prev));
}
}
/// Read the current gate value. `post_write_housekeeping` calls this
/// inside the `Some(src)` arm to decide whether to skip the
/// `push_mutation`.
pub(crate) fn is_applying_replicated() -> bool {
APPLYING_REPLICATED.with(Cell::get)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn default_is_off() {
assert!(!is_applying_replicated());
}
#[test]
fn guard_sets_then_clears() {
assert!(!is_applying_replicated());
{
let _g = ReplicatedApplyGuard::enter();
assert!(is_applying_replicated());
}
assert!(!is_applying_replicated());
}
#[test]
fn guard_nests_correctly() {
let _outer = ReplicatedApplyGuard::enter();
assert!(is_applying_replicated());
{
let _inner = ReplicatedApplyGuard::enter();
assert!(is_applying_replicated());
}
// Outer scope still active after inner drops.
assert!(is_applying_replicated());
}
}