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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
//! Startup diagnostics held until the effective output mode is known.
//!
//! Netsuke resolves its locale before it parses the command line, because
//! usage errors have to be rendered in the user's language. That ordering
//! creates a window: a locale that falls back to English is worth reporting,
//! but the JSON diagnostic document is written to stderr, and configuration can
//! still turn JSON on after the fallback has happened. Emitting immediately
//! risks corrupting that document; emitting at `OFF` loses the report.
//!
//! So startup events are written to a buffer instead of a stream. Once the
//! effective mode is settled, the buffer is either released to stderr (human
//! mode) or dropped (JSON mode), and everything after that is written straight
//! through.
use std::io::{self, Write};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use tracing_subscriber::fmt::MakeWriter;
/// The most startup diagnostics that will be held before the effective mode is
/// known.
///
/// The window is short — locale resolution, then command-line parsing — and
/// what it carries is a handful of one-line warnings. The bound exists so that
/// the size of what is buffered never depends on how much a run happens to
/// emit: without it, a pathological run could hold arbitrary bytes in memory
/// before anything decided where they belong.
///
/// Overflow keeps the first bytes rather than the last: the earliest
/// diagnostics describe how the run was configured, which is what a reader
/// needs, and later ones are progressively less informative about the startup
/// decision. Once full, [`TRUNCATION_MARKER`] is appended once, if it fits, and
/// everything after is dropped.
const MAX_BUFFERED_BYTES: usize = 64 * 1024;
/// Appended once when [`MAX_BUFFERED_BYTES`] is reached, so a truncated buffer
/// says so rather than appearing to be the whole of it.
const TRUNCATION_MARKER: &[u8] = b"\n[startup diagnostics truncated]\n";
/// Where startup diagnostics are going right now.
enum Sink {
/// Held until the effective mode is known.
Buffered(BoundedBuffer),
/// Human mode: written through to stderr.
Stderr,
/// JSON mode: discarded, so stderr carries only the diagnostic document.
Discard,
}
/// A byte buffer that stops growing at [`MAX_BUFFERED_BYTES`].
///
/// Once full it records that it truncated, so the marker is appended exactly
/// once however many writes follow.
#[derive(Default)]
struct BoundedBuffer {
/// Bytes held so far, bounded by `MAX_BUFFERED_BYTES`.
bytes: Vec<u8>,
/// Whether the bound was reached and the truncation marker appended.
truncated: bool,
}
impl BoundedBuffer {
/// Append as much of `buf` as the bound allows.
///
/// The first write to overflow keeps the bytes that fit, then appends the
/// truncation marker if there is room for it. Later writes are dropped.
fn append(&mut self, buf: &[u8]) {
if self.truncated {
return;
}
let remaining = MAX_BUFFERED_BYTES.saturating_sub(self.bytes.len());
if buf.len() <= remaining {
self.bytes.extend_from_slice(buf);
return;
}
// Room for the marker is reserved rather than claimed afterwards. A
// buffer filled to exactly the bound would leave none, and the marker
// would be dropped in precisely the case it is needed.
let content_limit = MAX_BUFFERED_BYTES.saturating_sub(TRUNCATION_MARKER.len());
if self.bytes.len() < content_limit {
let keep = content_limit.saturating_sub(self.bytes.len());
// `get` rather than a slice index: the lint forbids slicing that
// could panic, and a short `buf` is a legitimate input here.
if let Some(head) = buf.get(..keep.min(buf.len())) {
self.bytes.extend_from_slice(head);
}
}
self.bytes.truncate(content_limit);
self.bytes.extend_from_slice(TRUNCATION_MARKER);
self.truncated = true;
}
/// Drain the held bytes, resetting the truncation state.
fn take(&mut self) -> Vec<u8> {
self.truncated = false;
std::mem::take(&mut self.bytes)
}
#[cfg(test)]
fn as_slice(&self) -> &[u8] {
&self.bytes
}
}
/// A writer that buffers until told where the output belongs.
///
/// Cloned by the `fmt` layer for each event, so the sink is shared behind an
/// `Arc`; every clone observes the same state and the same buffer.
#[derive(Clone)]
pub struct StartupWriter {
/// Shared sink state visible to every clone of the writer.
sink: Arc<Mutex<Sink>>,
}
impl StartupWriter {
/// A writer that holds everything written to it.
///
/// # Examples
///
/// ```text
/// let w = StartupWriter::buffering();
/// warn!("locale fell back"); -> held; nothing reaches stderr
/// ```
///
/// Examples are shown rather than run: this module is compiled into the
/// binary, and Cargo does not run doctests for a binary target, so a
/// `rust` block would never be checked and would rot unnoticed.
#[must_use]
pub fn buffering() -> Self {
Self {
sink: Arc::new(Mutex::new(Sink::Buffered(BoundedBuffer::default()))),
}
}
/// Lock the shared sink, recovering from a poisoned mutex.
fn lock(&self) -> MutexGuard<'_, Sink> {
// A panic while formatting an event must not cascade into losing the
// rest of the diagnostics.
self.sink.lock().unwrap_or_else(PoisonError::into_inner)
}
/// Write everything buffered to stderr, and write through from now on.
///
/// Called when the effective mode turns out to be human.
///
/// # Examples
///
/// ```text
/// warn!("locale fell back"); -> held
/// w.release_to_stderr()?; -> the held bytes reach stderr, buffer emptied
/// warn!("something later"); -> written straight to stderr
/// ```
///
/// # Errors
///
/// Returns the error from writing the buffered bytes to stderr.
pub fn release_to_stderr(&self) -> io::Result<()> {
let mut sink = self.lock();
let buffered = match &mut *sink {
Sink::Buffered(buffer) => buffer.take(),
Sink::Stderr | Sink::Discard => Vec::new(),
};
*sink = Sink::Stderr;
drop(sink);
if buffered.is_empty() {
return Ok(());
}
io::stderr().write_all(&buffered)
}
/// Drop everything buffered, and discard whatever follows.
///
/// Called when the effective mode turns out to be JSON, so that stderr
/// carries only the diagnostic document.
///
/// # Examples
///
/// ```text
/// warn!("locale fell back"); -> held
/// w.discard(); -> the held bytes are dropped
/// warn!("something later"); -> dropped too, not re-buffered
/// ```
pub fn discard(&self) {
let mut sink = self.lock();
*sink = Sink::Discard;
}
/// The bytes currently held, for tests that assert what was recorded
/// before the mode was known.
///
/// Test-only: production code never inspects the buffer, it only decides
/// where the buffer goes.
#[cfg(test)]
#[must_use]
pub fn buffered(&self) -> Vec<u8> {
match &*self.lock() {
Sink::Buffered(buffer) => buffer.as_slice().to_vec(),
Sink::Stderr | Sink::Discard => Vec::new(),
}
}
}
/// The per-event handle the `fmt` layer writes through.
pub struct StartupWriterHandle {
/// Shared sink state this handle forwards events to.
sink: Arc<Mutex<Sink>>,
}
impl Write for StartupWriterHandle {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let mut sink = self.sink.lock().unwrap_or_else(PoisonError::into_inner);
match &mut *sink {
Sink::Buffered(buffer) => {
buffer.append(buf);
// The whole slice is reported as written even when the bound
// dropped some of it: the formatter has no recourse, and a
// short write would be reported through the channel being
// truncated.
Ok(buf.len())
}
Sink::Stderr => {
drop(sink);
io::stderr().write(buf)
}
// Report the bytes as written: the caller has no recourse, and a
// short-write error would be reported through the very channel
// being discarded.
Sink::Discard => Ok(buf.len()),
}
}
fn flush(&mut self) -> io::Result<()> {
let sink = self.sink.lock().unwrap_or_else(PoisonError::into_inner);
if matches!(&*sink, Sink::Stderr) {
drop(sink);
return io::stderr().flush();
}
Ok(())
}
}
impl<'writer> MakeWriter<'writer> for StartupWriter {
type Writer = StartupWriterHandle;
fn make_writer(&'writer self) -> Self::Writer {
StartupWriterHandle {
sink: Arc::clone(&self.sink),
}
}
}
#[cfg(test)]
#[path = "startup_tracing_tests.rs"]
mod tests;