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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
//! The console's counters.
//!
//! Relaxed atomics, each moved where its event happens and read at one moment
//! as [`ConsoleCountsSnapshot`]. Every console bookkeeping failure is one of
//! these, never a log line alone.
use std::sync::atomic::{AtomicU64, Ordering};
/// The live counters.
#[derive(Debug, Default)]
pub struct ConsoleCounts {
pub(super) topology_emit_failures: AtomicU64,
pub(super) topology_clock_failures: AtomicU64,
pub(super) station_collisions: AtomicU64,
pub(super) tap_observe_failures: AtomicU64,
pub(super) tap_attach_failures: AtomicU64,
pub(super) tap_detach_failures: AtomicU64,
pub(super) console_wake_failures: AtomicU64,
pub(super) console_publish_failures: AtomicU64,
pub(super) console_records_lost: AtomicU64,
pub(super) tap_records_abandoned: AtomicU64,
pub(super) tap_messages_abandoned: AtomicU64,
pub(super) console_pending_records: AtomicU64,
pub(super) topology_records_published: AtomicU64,
pub(super) tap_records_published: AtomicU64,
pub(super) console_pump_wakes: AtomicU64,
pub(super) taps_attached: AtomicU64,
pub(super) tap_suppressed_total: AtomicU64,
}
/// Every console counter, read at one moment.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ConsoleCountsSnapshot {
/// Topology changes the hub could not apply (a poisoned model lock).
pub topology_emit_failures: u64,
/// Topology documents sent with this device's start missing: the wall
/// clock gave no time for it, and the fault is named on the record.
pub topology_clock_failures: u64,
/// Worker identities or service subjects left off the map because they
/// are this server's exchange or depot id: kept apart, never merged
/// into those stations (Buttercup's V1, Hermes 04:27).
pub station_collisions: u64,
/// Tap observe failures: a publish the tap could not observe (a
/// poisoned tap lock), a station list refused, or a hop that could not
/// be kept as measured; each failure of one observe counts.
pub tap_observe_failures: u64,
/// Tap subscriptions that could not start.
pub tap_attach_failures: u64,
/// Tap subscriptions that could not stop.
pub tap_detach_failures: u64,
/// Wakes that reached no publisher (a poisoned notifier slot).
pub console_wake_failures: u64,
/// Drops of a tap subscription that could not leave the subscriber list.
pub tap_unsubscribe_failures: u64,
/// Console records whose publication failed (each kept and retried).
pub console_publish_failures: u64,
/// Drained console records thrown away. The ring is the only buffer and a
/// tap with a pending batch is not drained, so this stays 0: a control
/// against a second buffer creeping back in.
pub console_records_lost: u64,
/// Encoded tap records left unsent when their tap went: its last
/// subscriber left, or it was attached anew before they were sent.
pub tap_records_abandoned: u64,
/// Messages still in a tap's ring when its last subscriber left.
pub tap_messages_abandoned: u64,
/// Console records held for retry after a failed publication.
pub console_pending_records: u64,
/// `console.topology` records published.
pub topology_records_published: u64,
/// `console.tap.<target>` records published.
pub tap_records_published: u64,
/// Times the console publisher was woken and ran.
pub console_pump_wakes: u64,
/// Taps live now.
pub taps_attached: u64,
/// Messages evicted from every tap's ring since start.
pub tap_dropped_total: u64,
/// Dedup-suppressed re-publishes, which are not tapped.
pub tap_suppressed_total: u64,
}
/// A console bookkeeping failure, by name (SRV-012 R8). Each is a counter
/// here and the `kind` of `liminal_console_failures_total`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ConsoleFailure {
/// A topology change the model could not take.
TopologyEmit,
/// A publish the tap could not observe.
TapObserve,
/// A tap subscription that could not start.
TapAttach,
/// A tap subscription that could not stop.
TapDetach,
/// A wake that reached no publisher.
ConsoleWake,
/// A dropped tap subscription that could not leave the list.
TapUnsubscribe,
/// A console record whose publication failed.
ConsolePublish,
/// A drained console record thrown away.
ConsoleRecordsLost,
/// A topology document whose start the wall clock could not give.
TopologyClock,
/// A worker identity or service subject equal to this server's exchange
/// or depot id, left off the map.
StationCollision,
}
impl ConsoleFailure {
/// The metric label values, in [`Self::slot`] order.
pub const LABELS: [&'static str; 10] = [
"topology_emit",
"tap_observe",
"tap_attach",
"tap_detach",
"console_wake",
"tap_unsubscribe",
"console_publish",
"console_records_lost",
"topology_clock",
"station_collision",
];
/// This failure's index into [`Self::LABELS`].
#[must_use]
pub const fn slot(self) -> usize {
match self {
Self::TopologyEmit => 0,
Self::TapObserve => 1,
Self::TapAttach => 2,
Self::TapDetach => 3,
Self::ConsoleWake => 4,
Self::TapUnsubscribe => 5,
Self::ConsolePublish => 6,
Self::ConsoleRecordsLost => 7,
Self::TopologyClock => 8,
Self::StationCollision => 9,
}
}
}
/// Counts `amount` failures of `kind` in `counter` and in its metric.
pub(crate) fn failed(counter: &AtomicU64, amount: u64, kind: ConsoleFailure) {
add(counter, amount);
crate::metrics::console_failure(kind, amount);
}
pub(super) fn add(counter: &AtomicU64, amount: u64) {
counter.fetch_add(amount, Ordering::Relaxed);
}
pub(super) fn set(counter: &AtomicU64, value: u64) {
counter.store(value, Ordering::Relaxed);
}
impl ConsoleCounts {
/// Every counter as it stands. `tap_dropped_total` and
/// `tap_unsubscribe_failures` are the tap's own, passed in.
#[must_use]
pub fn snapshot(
&self,
tap_dropped_total: u64,
tap_unsubscribe_failures: u64,
) -> ConsoleCountsSnapshot {
let read = |counter: &AtomicU64| counter.load(Ordering::Relaxed);
ConsoleCountsSnapshot {
topology_emit_failures: read(&self.topology_emit_failures),
topology_clock_failures: read(&self.topology_clock_failures),
station_collisions: read(&self.station_collisions),
tap_observe_failures: read(&self.tap_observe_failures),
tap_attach_failures: read(&self.tap_attach_failures),
tap_detach_failures: read(&self.tap_detach_failures),
console_wake_failures: read(&self.console_wake_failures),
tap_unsubscribe_failures,
console_publish_failures: read(&self.console_publish_failures),
console_records_lost: read(&self.console_records_lost),
tap_records_abandoned: read(&self.tap_records_abandoned),
tap_messages_abandoned: read(&self.tap_messages_abandoned),
console_pending_records: read(&self.console_pending_records),
topology_records_published: read(&self.topology_records_published),
tap_records_published: read(&self.tap_records_published),
console_pump_wakes: read(&self.console_pump_wakes),
taps_attached: read(&self.taps_attached),
tap_dropped_total,
tap_suppressed_total: read(&self.tap_suppressed_total),
}
}
}