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
//! Glue between the public `Store` surface and the replica runner —
//! the constructor that decides whether to spawn the background
//! thread, plus the read-only enforcement helper that every mutating
//! API in `ops.rs` calls. Extracted from `store.rs` to keep that file
//! under the 500-LOC ceiling.
use std::sync::atomic::{AtomicU64, Ordering};
use crate::config::Config;
use crate::store::Shards;
/// Construct + spawn the replica runner when configured. Returns
/// `None` when the upstream is unset (normal primary store). The
/// returned handle is owned by `DropGuard`, which joins the runner
/// thread on the last `Store` clone drop.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn spawn_replica_runner(
config: &Config,
shards: &Shards,
) -> Option<crate::replica_runner::ReplicaRunner> {
let upstream = config.replica_upstream.as_ref()?.clone();
Some(crate::replica_runner::ReplicaRunner::spawn(
shards.clone(),
upstream,
config.replica_id.clone(),
config.replica_reconnect_min,
config.replica_reconnect_max,
))
}
/// Generate a process-unique replica id for `Store::open_replica`.
/// Format: `"kevy-embedded-{pid}-{seq}"` — the pid stays stable across
/// a process, the seq counter advances per open so two embeds in the
/// same process don't collide on a single slot in the primary's
/// SlotTable (the bug that would otherwise cause backlog frames a
/// fresh replica still needs to be evicted as soon as the prior
/// embed disconnects).
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn fresh_replica_id() -> String {
static SEQ: AtomicU64 = AtomicU64::new(0);
let n = SEQ.fetch_add(1, Ordering::Relaxed);
format!("kevy-embedded-{}-{}", std::process::id(), n)
}
impl crate::Store {
/// The embed-writer replication listener's actually-bound address,
/// when this store was opened with
/// [`Config::with_embed_writer`](crate::Config::with_embed_writer).
/// Open with port `0` to let the OS pick a free port and read the
/// real one back here — the race-free way to run an ephemeral
/// writer (tests, sidecars, one process per scope).
#[cfg(not(target_arch = "wasm32"))]
pub fn writer_addr(&self) -> Option<std::net::SocketAddr> {
self.guard.replica_source.as_ref().map(|s| s.local_addr())
}
}