log_io/
sink.rs

1//! Output sinks.
2//!
3//! A sink takes a [`Record`], serializes it with its [`Format`] and
4//! delivers the bytes somewhere. The built-in sinks are:
5//!
6//! | Sink | Destination | Notes |
7//! |---|---|---|
8//! | [`StdoutSink`], [`StderrSink`] | standard streams | written under the stream's own lock |
9//! | [`FileSink`] | a file | buffered; see [`FileSink`] for durability |
10//! | [`WriterSink`] | any `io::Write` | e.g. a [`RotatingFile`] ([`RotatingFileSink`]) |
11//! | [`NonBlocking`] | any `io::Write`, on a background thread | bounded queue, explicit overflow policy |
12//! | [`NullSink`] | nowhere | benchmarks and tests |
13//!
14//! # Threading
15//!
16//! Every sink is `Send + Sync`. The synchronous sinks serialize writers
17//! with a mutex and perform the write on the calling thread, so in an
18//! async runtime they block the executor thread for the duration of
19//! the write. Use [`NonBlocking`] to move I/O off the calling thread.
20//!
21//! # Allocation
22//!
23//! Records are formatted into a reusable thread-local buffer, so the
24//! synchronous sinks do not allocate in steady state. A buffer that
25//! grew past 64 KiB for an unusually large record is shrunk back to
26//! 64 KiB afterwards, so one huge record does not pin memory on every
27//! thread for the life of the process.
28//!
29//! # Reliability
30//!
31//! The built-in sinks recover from mutex poisoning: if a thread panics
32//! while holding a sink's lock (for example inside a custom writer),
33//! later records are still written. Logging stays available during
34//! incident response.
35
36use core::cell::RefCell;
37use std::io::{self, Write};
38use std::sync::{Mutex, MutexGuard};
39
40use crate::error::{Error, Result};
41use crate::format::Format;
42use crate::record::Record;
43
44/// Output stage of the pipeline.
45///
46/// Implementors must be safe to call from many threads concurrently.
47/// A sink should never panic on bad input; return an error instead so
48/// the logger can report it and keep feeding the other sinks.
49///
50/// # Examples
51///
52#[cfg_attr(all(feature = "json"), doc = "```")]
53#[cfg_attr(not(all(feature = "json")), doc = "```ignore")]
54/// use std::sync::Mutex;
55/// use log_io::format::{Format, JsonFormat};
56/// use log_io::{Level, Logger, Record, Result, Sink};
57///
58/// /// Collects formatted records in memory.
59/// #[derive(Default)]
60/// struct Memory(Mutex<Vec<String>>);
61///
62/// impl Sink for Memory {
63///     fn write_record(&self, record: &Record<'_>) -> Result<()> {
64///         let mut line = String::new();
65///         JsonFormat::new().write_record(record, &mut line)?;
66///         self.0.lock().unwrap_or_else(|p| p.into_inner()).push(line);
67///         Ok(())
68///     }
69///     fn flush(&self) -> Result<()> {
70///         Ok(())
71///     }
72/// }
73///
74/// let sink = std::sync::Arc::new(Memory::default());
75/// let logger = Logger::builder().no_timestamps().with_sink(sink.clone()).build();
76/// logger.log(Level::Info, "hello", &[]);
77/// assert_eq!(sink.0.lock().unwrap().len(), 1);
78/// ```
79pub trait Sink: Send + Sync {
80    /// Serialize and deliver one record.
81    ///
82    /// # Errors
83    ///
84    /// Returns the destination's I/O error, a format error, or a
85    /// sink-specific error.
86    fn write_record(&self, record: &Record<'_>) -> Result<()>;
87
88    /// Push any buffered bytes to the destination (for files, to the
89    /// operating system; see the sink's own docs for `fsync`).
90    ///
91    /// # Errors
92    ///
93    /// Returns the destination's I/O error, or an error deferred from
94    /// an earlier background operation.
95    fn flush(&self) -> Result<()>;
96
97    /// Flush triggered by a record at or above the logger's flush level
98    /// (see [`crate::LoggerBuilder::flush_on`]).
99    ///
100    /// The default implementation calls [`Sink::flush`], so the record
101    /// has reached the destination when the logging call returns. A
102    /// background sink may override it to hand the request to its
103    /// worker without waiting, when waiting would break its own latency
104    /// contract; [`NonBlocking`] does this under
105    /// [`Overflow::DropNewest`]. [`crate::Logger::flush`] and
106    /// [`crate::Logger::shutdown`] call [`Sink::flush`] and
107    /// [`Sink::shutdown`], never this method.
108    ///
109    /// # Errors
110    ///
111    /// Same as [`Sink::flush`]. A sink that does not wait can only
112    /// report errors it already knows about.
113    fn request_flush(&self) -> Result<()> {
114        self.flush()
115    }
116
117    /// Flush and release background resources. Called by
118    /// [`crate::Logger::shutdown`], by a [`crate::FlushGuard`] on drop,
119    /// and when the last clone of a [`crate::Logger`] is dropped.
120    ///
121    /// A sink must keep accepting records after `shutdown` (writing
122    /// them synchronously if its background machinery is gone): late
123    /// records from other threads must not be lost. The default
124    /// implementation calls [`Sink::flush`].
125    ///
126    /// # Errors
127    ///
128    /// Same as [`Sink::flush`].
129    fn shutdown(&self) -> Result<()> {
130        self.flush()
131    }
132}
133
134impl<S: Sink + ?Sized> Sink for std::sync::Arc<S> {
135    fn write_record(&self, record: &Record<'_>) -> Result<()> {
136        (**self).write_record(record)
137    }
138    fn flush(&self) -> Result<()> {
139        (**self).flush()
140    }
141    fn request_flush(&self) -> Result<()> {
142        (**self).request_flush()
143    }
144    fn shutdown(&self) -> Result<()> {
145        (**self).shutdown()
146    }
147}
148
149impl<S: Sink + ?Sized> Sink for Box<S> {
150    fn write_record(&self, record: &Record<'_>) -> Result<()> {
151        (**self).write_record(record)
152    }
153    fn flush(&self) -> Result<()> {
154        (**self).flush()
155    }
156    fn request_flush(&self) -> Result<()> {
157        (**self).request_flush()
158    }
159    fn shutdown(&self) -> Result<()> {
160        (**self).shutdown()
161    }
162}
163
164mod file;
165mod non_blocking;
166mod recent;
167mod rotating;
168mod stdio;
169mod writer;
170
171pub use self::file::{FileSink, FileSinkBuilder};
172pub(crate) use self::non_blocking::on_background_worker;
173pub use self::non_blocking::{NonBlocking, NonBlockingBuilder, Overflow};
174pub use self::recent::{DEFAULT_LINE_LIMIT, RecentLines};
175pub use self::rotating::{RotatingFile, RotatingFileBuilder, Rotation};
176pub use self::stdio::{StderrSink, StdoutSink};
177pub use self::writer::{NullSink, WriterSink};
178
179/// A [`WriterSink`] over a [`RotatingFile`].
180///
181/// # Examples
182///
183#[cfg_attr(all(feature = "json"), doc = "```no_run")]
184#[cfg_attr(not(all(feature = "json")), doc = "```ignore")]
185/// use log_io::format::JsonFormat;
186/// use log_io::sink::{RotatingFile, RotatingFileSink, Rotation};
187///
188/// let file = RotatingFile::builder("logs/app.jsonl")
189///     .max_bytes(10 * 1024 * 1024)
190///     .rotation(Rotation::Daily)
191///     .max_files(7)
192///     .open()?;
193/// let sink: RotatingFileSink<JsonFormat> = RotatingFileSink::new(file, JsonFormat::new());
194/// # Ok::<(), std::io::Error>(())
195/// ```
196pub type RotatingFileSink<F> = WriterSink<RotatingFile, F>;
197
198/// Keeps a sink's lock on its own cache lines.
199///
200/// Every logging thread reads the logger's sink table on each record,
201/// and every thread writes the sink's mutex. If a small sink allocation
202/// shares a cache line with other hot data, each lock and unlock
203/// invalidates that line for all threads. Measured on the reference
204/// machine, padding restored the 4-thread throughput of 1.0.
205#[repr(align(128))]
206pub(crate) struct CachePadded<T>(pub(crate) T);
207
208impl<T> core::ops::Deref for CachePadded<T> {
209    type Target = T;
210    fn deref(&self) -> &T {
211        &self.0
212    }
213}
214
215/// Capacity a thread-local format buffer may keep between records.
216pub(crate) const MAX_RETAINED_CAPACITY: usize = 64 * 1024;
217
218thread_local! {
219    /// Reusable formatting buffer, so steady-state formatting does not
220    /// touch the allocator.
221    static FORMAT_BUF: RefCell<String> = const { RefCell::new(String::new()) };
222}
223
224/// Shrink `buf` back to [`MAX_RETAINED_CAPACITY`] after an oversized
225/// record. Cheap when no shrinking is needed: one comparison.
226#[inline]
227pub(crate) fn trim_buffer(buf: &mut String) {
228    if buf.capacity() > MAX_RETAINED_CAPACITY {
229        buf.clear();
230        buf.shrink_to(MAX_RETAINED_CAPACITY);
231    }
232}
233
234/// Format `record` into the thread-local buffer and pass the text to
235/// `write`.
236///
237/// If the buffer is already in use on this thread (a sink that logs
238/// from inside its own write) or thread-local storage is being torn
239/// down, a temporary buffer is used instead.
240pub(crate) fn with_formatted<F, W>(format: &F, record: &Record<'_>, mut write: W) -> Result<()>
241where
242    F: Format,
243    W: FnMut(&str) -> io::Result<()>,
244{
245    let pooled = FORMAT_BUF.try_with(|cell| {
246        let mut buf = cell.try_borrow_mut().ok()?;
247        Some(format_into(&mut buf, format, record, &mut write))
248    });
249    match pooled {
250        Ok(Some(result)) => result,
251        _ => format_into(&mut String::with_capacity(256), format, record, &mut write),
252    }
253}
254
255fn format_into<F, W>(buf: &mut String, format: &F, record: &Record<'_>, write: &mut W) -> Result<()>
256where
257    F: Format,
258    W: FnMut(&str) -> io::Result<()>,
259{
260    buf.clear();
261    let result = format
262        .write_record_generic(record, buf)
263        .map_err(Error::Format)
264        .and_then(|()| write(buf).map_err(Error::Io));
265    trim_buffer(buf);
266    result
267}
268
269/// Unix permission bits for a log file created without an explicit
270/// `mode`: owner read and write only (since 2.2; `0o666` minus the umask
271/// before).
272pub(crate) const DEFAULT_FILE_MODE: u32 = 0o600;
273
274/// Applies a Unix permission mode to files `options` creates. On other
275/// platforms the file inherits its directory's access control list and
276/// `mode` has no effect.
277pub(crate) fn apply_mode(options: &mut std::fs::OpenOptions, mode: u32) {
278    #[cfg(unix)]
279    {
280        use std::os::unix::fs::OpenOptionsExt;
281        let _options = options.mode(mode);
282    }
283    #[cfg(not(unix))]
284    let _unused = (options, mode);
285}
286
287/// `O_NOFOLLOW` for the target, from the platform headers (the values the
288/// `libc` crate uses). `None` on Unix systems not listed: the checks in
289/// [`open_log_file`] still refuse a symbolic link there, but only after
290/// `open` has followed it.
291#[cfg(any(target_os = "linux", target_os = "android"))]
292const O_NOFOLLOW: Option<i32> = Some(
293    if cfg!(any(
294        target_arch = "arm",
295        target_arch = "aarch64",
296        target_arch = "powerpc",
297        target_arch = "powerpc64",
298        target_arch = "m68k"
299    )) {
300        0x8000
301    } else {
302        0x2_0000
303    },
304);
305#[cfg(any(
306    target_vendor = "apple",
307    target_os = "freebsd",
308    target_os = "netbsd",
309    target_os = "openbsd",
310    target_os = "dragonfly"
311))]
312const O_NOFOLLOW: Option<i32> = Some(0x100);
313#[cfg(any(target_os = "solaris", target_os = "illumos"))]
314const O_NOFOLLOW: Option<i32> = Some(0x2_0000);
315#[cfg(all(
316    unix,
317    not(any(
318        target_os = "linux",
319        target_os = "android",
320        target_vendor = "apple",
321        target_os = "freebsd",
322        target_os = "netbsd",
323        target_os = "openbsd",
324        target_os = "dragonfly",
325        target_os = "solaris",
326        target_os = "illumos"
327    ))
328))]
329const O_NOFOLLOW: Option<i32> = None;
330
331fn refuse(message: &'static str) -> io::Error {
332    io::Error::new(io::ErrorKind::InvalidInput, message)
333}
334
335/// Opens (creating if needed) a log file for writing.
336///
337/// Unless `follow_symlinks` is set, the path must name a regular file,
338/// not a symbolic link (or, on Windows, another reparse point): a link
339/// planted in a shared log directory must not redirect, or truncate,
340/// another file. On Unix the open uses `O_NOFOLLOW` where the platform
341/// has it; everywhere the opened handle is then checked to be a regular
342/// file that is still the one at `path`. Truncation (`append == false`)
343/// happens only after those checks.
344///
345/// New files get `mode` (Unix), [`DEFAULT_FILE_MODE`] when `None`.
346pub(crate) fn open_log_file(
347    path: &std::path::Path,
348    append: bool,
349    mode: Option<u32>,
350    follow_symlinks: bool,
351) -> io::Result<std::fs::File> {
352    let mut options = std::fs::OpenOptions::new();
353    let _options = options.create(true).write(true).append(append);
354    apply_mode(&mut options, mode.unwrap_or(DEFAULT_FILE_MODE));
355    if !follow_symlinks {
356        if std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_symlink()) {
357            return Err(refuse(
358                "log file path is a symbolic link; set follow_symlinks(true) to allow it",
359            ));
360        }
361        #[cfg(unix)]
362        if let Some(flag) = O_NOFOLLOW {
363            use std::os::unix::fs::OpenOptionsExt;
364            let _options = options.custom_flags(flag);
365        }
366        #[cfg(windows)]
367        {
368            use std::os::windows::fs::OpenOptionsExt;
369            // Opens a reparse point itself instead of its target.
370            const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
371            let _options = options.custom_flags(FILE_FLAG_OPEN_REPARSE_POINT);
372        }
373    }
374    let file = options.open(path)?;
375    if !follow_symlinks {
376        let opened = file.metadata()?;
377        if !opened.file_type().is_file() {
378            return Err(refuse("log file path is not a regular file"));
379        }
380        let named = std::fs::symlink_metadata(path)?;
381        if named.file_type().is_symlink() {
382            return Err(refuse(
383                "log file path is a symbolic link; set follow_symlinks(true) to allow it",
384            ));
385        }
386        #[cfg(unix)]
387        {
388            use std::os::unix::fs::MetadataExt;
389            if (named.dev(), named.ino()) != (opened.dev(), opened.ino()) {
390                return Err(refuse(
391                    "log file path was replaced while it was being opened",
392                ));
393            }
394        }
395    }
396    if !append {
397        file.set_len(0)?;
398    }
399    Ok(file)
400}
401
402/// Acquire a mutex guard even if the mutex is poisoned.
403///
404/// Poisoning only means a previous holder unwound; the writer inside is
405/// still usable, and logging must keep working after a panic.
406pub(crate) fn lock_recover<T: ?Sized>(m: &Mutex<T>) -> MutexGuard<'_, T> {
407    m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
408}
409
410/// Format `record` and write it to the writer behind `target`.
411pub(crate) fn write_locked<W: Write + ?Sized, F: Format>(
412    target: &Mutex<W>,
413    format: &F,
414    record: &Record<'_>,
415) -> Result<()> {
416    with_formatted(format, record, |text| {
417        lock_recover(target).write_all(text.as_bytes())
418    })
419}
420
421/// Flush the writer behind `target`.
422pub(crate) fn flush_locked<W: Write + ?Sized>(target: &Mutex<W>) -> Result<()> {
423    lock_recover(target).flush().map_err(Error::from)
424}
425
426#[cfg(all(test, feature = "json"))]
427mod tests {
428    use super::*;
429    use crate::format::JsonFormat;
430    use crate::{Level, Metadata};
431
432    #[test]
433    fn test_trim_buffer_caps_retained_capacity() {
434        let mut buf = String::with_capacity(4 * MAX_RETAINED_CAPACITY);
435        buf.push('x');
436        trim_buffer(&mut buf);
437        assert!(buf.capacity() <= MAX_RETAINED_CAPACITY);
438        let mut small = String::with_capacity(128);
439        small.push('y');
440        trim_buffer(&mut small);
441        assert_eq!(small, "y", "small buffers are left alone");
442    }
443
444    #[test]
445    fn test_thread_local_buffer_shrinks_after_oversize_record() {
446        // The audit's LOG-007 reproduction, scaled down: one 1 MiB
447        // record must not leave a 1 MiB+ buffer behind.
448        std::thread::spawn(|| {
449            let big = "x".repeat(1024 * 1024);
450            let record = Record::new(Metadata::new(Level::Info, "t"), &big, &[]);
451            let mut written = 0;
452            with_formatted(&JsonFormat::new(), &record, |s| {
453                written = s.len();
454                Ok(())
455            })
456            .unwrap();
457            assert!(written > big.len());
458            let cap = FORMAT_BUF.with(|b| b.borrow().capacity());
459            assert!(cap <= MAX_RETAINED_CAPACITY, "retained {cap} bytes");
460        })
461        .join()
462        .unwrap();
463    }
464
465    #[test]
466    fn test_with_formatted_reentrant_call_uses_temporary_buffer() {
467        let record = Record::new(Metadata::new(Level::Info, "t"), "outer", &[]);
468        let mut inner_text = String::new();
469        with_formatted(&JsonFormat::new(), &record, |outer| {
470            let inner = Record::new(Metadata::new(Level::Info, "t"), "inner", &[]);
471            with_formatted(&JsonFormat::new(), &inner, |s| {
472                inner_text = s.to_owned();
473                Ok(())
474            })
475            .map_err(|e| io::Error::other(e.to_string()))?;
476            assert!(outer.contains("outer"));
477            Ok(())
478        })
479        .unwrap();
480        assert!(inner_text.contains("inner"));
481    }
482
483    #[test]
484    fn test_with_formatted_surfaces_write_and_format_errors() {
485        let record = Record::new(Metadata::new(Level::Info, "t"), "m", &[]);
486        let err = with_formatted(&JsonFormat::new(), &record, |_| Err(io::Error::other("x")))
487            .unwrap_err();
488        assert!(matches!(err, Error::Io(_)));
489
490        struct Broken;
491        impl core::fmt::Display for Broken {
492            fn fmt(&self, _: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
493                Err(core::fmt::Error)
494            }
495        }
496        let fields = [crate::Field::new("b", crate::Value::Display(&Broken))];
497        let record = Record::new(Metadata::new(Level::Info, "t"), "m", &fields);
498        let err = with_formatted(&JsonFormat::new(), &record, |_| Ok(())).unwrap_err();
499        assert!(matches!(err, Error::Format(_)));
500    }
501}
log_io/sink.rs

log_io/
sink.rs
