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
use super::{
Arc, AtomicBool, ConnectionControl, ConnectionIncarnation, HashMap, MountKind, Mutex,
MutexGuard, PassPrincipal, RawFd, ReadinessToken, ServerError, SocketAddr, TcpStream,
WorkerRegistration,
};
/// The connection records and the service directory built from them, under
/// one lock (SRV-027 R1): a directory entry is added when its record stores a
/// registration and removed in the same lock take that removes or replaces
/// the record, so no lookup names a connection whose record is gone.
#[derive(Debug, Default)]
pub(super) struct Registry {
pub(super) connections: HashMap<u64, ConnectionRecord>,
pub(super) directory: super::service_directory::ServiceDirectory,
}
#[derive(Debug)]
pub(super) struct ConnectionRecord {
pub(super) peer_addr: Option<SocketAddr>,
/// Which door admitted this connection (design §10). Written once, by the
/// spawn path, from the server's own knowledge of which path it is; never
/// read from, or influenced by, anything the client sends.
pub(super) mount: MountKind,
/// Durable pair allocated and flushed before the process was spawned.
pub(super) connection_incarnation: Option<ConnectionIncarnation>,
/// Worker registration declared on this connection, set by `set_registration`
/// when a `WorkerRegister` frame is accepted. `Some` marks a connection whose
/// close must fire `on_worker_unregistered`.
pub(super) registration: Option<WorkerRegistration>,
/// Registry principal a pass-stamped `Connect` verified, set by
/// `attach_pass_principal` once the handshake succeeded. `Some` marks a
/// connection whose close must fire `on_pass_detached` with it.
pub(super) pass_principal: Option<PassPrincipal>,
/// Host-reachable identity for ACK'd deregistration after external death.
pub(super) readiness: Option<ReadinessRegistration>,
/// Shared edge set before READY enters beamr's process-table pending queue;
/// the executing process reads it at its final probe.
pub(super) ready_pending: Arc<AtomicBool>,
/// FIX A-ii: `true` while this connection is parked (its last slice returned
/// `Wait`). A parked connection with no `ready_pending` edge has fanned out
/// every accepted publish to its socket, so the shutdown flush barrier reads
/// this with `ready_pending` to know delivery has quiesced before it lets the
/// shutdown Disconnect be broadcast.
pub(super) parked: AtomicBool,
/// Keeps the fd alive until deregistration has been acknowledged, preventing
/// stale registration delivery to a subsequently reused descriptor number.
pub(super) fd_guard: Option<TcpStream>,
}
#[derive(Debug, Clone, Copy)]
pub(super) struct ReadinessRegistration {
pub(super) token: ReadinessToken,
pub(super) fd: RawFd,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct QueuedConnectionControl {
pub(super) pid: u64,
pub(super) control: ConnectionControl,
}
pub(super) fn lock<'a, T>(
mutex: &'a Mutex<T>,
context: &str,
) -> Result<MutexGuard<'a, T>, ServerError> {
mutex.lock().map_err(|error| ServerError::ListenerAccept {
message: format!("{context} unavailable: {error}"),
})
}
/// Locks `mutex`, RECOVERING a poisoned guard instead of failing (S4). For
/// lifecycle-cleanup paths only (reply delivery, expiry, cancellation, the
/// close sweep): removal-style operations are sound on a recovered map, and a
/// cleanup that silently skipped its removal would strand slots and their §5
/// cap admissions forever. Admission paths keep the fail-closed [`lock`].
pub(super) fn recover_lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}