journal_log_writer/log/
mod.rs

1mod chain;
2use chain::OwnedChain;
3
4mod config;
5pub use config::{Config, LogIdentityMode, LogOpenMode, RetentionPolicy, RotationPolicy};
6
7mod helpers;
8mod startup;
9use helpers::*;
10use startup::{ActiveFile, RotationState, build_startup_state};
11
12use crate::{Result, WriterError};
13use itoa::Buffer as ItoaBuffer;
14pub use journal_common::EntryTimestamps;
15use journal_common::{Microseconds, RealtimeClock};
16use journal_core::error::JournalError;
17use journal_core::file::mmap::MmapMut;
18use journal_core::file::{
19    Compression, EntryField, EntryWriteOptions, FieldNamePolicy, JournalFile, JournalFileOptions,
20    JournalWriter, StructuredField,
21};
22use journal_registry::repository;
23use std::path::{Path, PathBuf};
24use std::sync::Arc;
25#[cfg(test)]
26use std::sync::atomic::{AtomicUsize, Ordering};
27
28const STACK_ENTRY_REF_LIMIT: usize = 128;
29const SOURCE_REALTIME_PREFIX: &[u8] = b"_SOURCE_REALTIME_TIMESTAMP=";
30const DERIVED_ROTATION_FRACTION: u64 = 20;
31const JOURNAL_FILE_SIZE_MIN: u64 = 512 * 1024;
32const PAGE_SIZE: u64 = 4096;
33const JOURNAL_COMPACT_SIZE_MAX: u64 = u32::MAX as u64;
34
35#[cfg(test)]
36static ARCHIVE_SYNC_CALLS: AtomicUsize = AtomicUsize::new(0);
37
38fn sync_archive_journal_file(
39    sync_on_archive: bool,
40    journal_file: &mut JournalFile<MmapMut>,
41) -> Result<()> {
42    if !sync_on_archive {
43        return Ok(());
44    }
45    #[cfg(test)]
46    ARCHIVE_SYNC_CALLS.fetch_add(1, Ordering::Relaxed);
47    journal_file.sync()?;
48    Ok(())
49}
50
51/// Tracks rotation state for size and count limits.
52pub struct Log {
53    configured_dir: PathBuf,
54    chain: OwnedChain,
55    config: Config,
56    active_file: Option<ActiveFile>,
57    rotation_state: RotationState,
58    boot_id: uuid::Uuid,
59    seqnum_id: uuid::Uuid,
60    current_seqnum: u64,
61    clock: RealtimeClock,
62    last_monotonic_usec: u64,
63    lifecycle_observer: Option<Arc<dyn LogLifecycleObserver>>,
64    artifact_sizer: Option<Arc<dyn LogArtifactSizer>>,
65    retention_on_open_applied: bool,
66    boot_id_field: Vec<u8>,
67    source_realtime_field: Vec<u8>,
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71pub enum LogLifecycleReason {
72    Append,
73    EagerOpen,
74    Rotation,
75    Retention,
76}
77
78#[derive(Debug, Clone)]
79pub enum LogLifecycleEvent {
80    Created {
81        active: repository::File,
82        reason: LogLifecycleReason,
83    },
84    Rotated {
85        archived: repository::File,
86        active: repository::File,
87    },
88    RetainedDeleted {
89        files: Vec<repository::File>,
90    },
91}
92
93pub trait LogLifecycleObserver: Send + Sync {
94    fn on_event(&self, event: &LogLifecycleEvent);
95}
96
97pub trait LogArtifactSizer: Send + Sync {
98    fn journal_artifact_size(&self, journal_path: &Path) -> Result<u64>;
99}
100
101impl Log {
102    fn duration_to_micros(duration: std::time::Duration) -> u64 {
103        duration.as_micros().try_into().unwrap_or(u64::MAX)
104    }
105
106    fn peek_entry_realtime(&self, timestamps: &EntryTimestamps) -> u64 {
107        let candidate = timestamps
108            .entry_realtime_usec
109            .unwrap_or_else(|| Microseconds::now().get());
110        let last_seen = self.clock.last_seen().get();
111        if candidate > last_seen {
112            candidate
113        } else {
114            last_seen.saturating_add(1)
115        }
116    }
117
118    fn should_rotate_for_realtime(&self, realtime: u64) -> bool {
119        let Some(active_file) = &self.active_file else {
120            return true;
121        };
122        if self.rotation_state.should_rotate() {
123            return true;
124        }
125        let Some(max_duration) = self.config.rotation_policy.duration_of_journal_file else {
126            return false;
127        };
128        let header = active_file.journal_file.journal_header_ref();
129        header.n_entries > 0
130            && header.head_entry_realtime > 0
131            && realtime.saturating_sub(header.head_entry_realtime)
132                >= Self::duration_to_micros(max_duration)
133    }
134
135    fn append_rotation_reason(&self) -> LogLifecycleReason {
136        if self.active_file.is_none() {
137            LogLifecycleReason::Append
138        } else {
139            LogLifecycleReason::Rotation
140        }
141    }
142
143    fn prepare_append_for_realtime(&mut self, entry_realtime: u64) -> Result<()> {
144        self.apply_retention_on_open()?;
145        let opened_first_active = self.active_file.is_none();
146        if self.should_rotate_for_realtime(entry_realtime) {
147            self.rotate(entry_realtime, self.append_rotation_reason())?;
148            if opened_first_active {
149                self.retention_on_open_applied = true;
150            }
151        }
152        self.apply_retention_on_open()
153    }
154
155    fn raw_items_for_policy<'a>(&self, items: &'a [&'a [u8]]) -> Result<Option<Vec<&'a [u8]>>> {
156        if self.config.field_name_policy != FieldNamePolicy::JournalApp {
157            return Ok(None);
158        }
159        let filtered_items = filter_raw_items_for_journal_app(items)?;
160        if filtered_items.is_empty() {
161            return Err(WriterError::EmptyEntry);
162        }
163        Ok(Some(filtered_items))
164    }
165
166    fn structured_fields_for_policy<'a>(
167        &self,
168        fields: &'a [StructuredField<'a>],
169    ) -> Result<Option<Vec<StructuredField<'a>>>> {
170        if self.config.field_name_policy != FieldNamePolicy::JournalApp {
171            return Ok(None);
172        }
173        let filtered_fields = filter_structured_fields_for_journal_app(fields);
174        if filtered_fields.is_empty() {
175            return Err(WriterError::EmptyEntry);
176        }
177        Ok(Some(filtered_fields))
178    }
179
180    fn apply_retention(&mut self, protected_file: Option<&repository::File>) -> Result<()> {
181        if let Some(sizer) = &self.artifact_sizer {
182            self.chain.refresh_retained_sizes(|file| {
183                sizer.journal_artifact_size(Path::new(file.path()))
184            })?;
185        }
186        let retention = self
187            .chain
188            .retain(&self.config.retention_policy, protected_file);
189        let deleted_files = retention.deleted_files;
190        if !deleted_files.is_empty()
191            && let Some(observer) = &self.lifecycle_observer
192        {
193            observer.on_event(&LogLifecycleEvent::RetainedDeleted {
194                files: deleted_files,
195            });
196        }
197        if let Some(error) = retention.error {
198            return Err(error);
199        }
200
201        Ok(())
202    }
203
204    fn apply_retention_on_open(&mut self) -> Result<()> {
205        if self.retention_on_open_applied || self.active_file.is_none() {
206            return Ok(());
207        }
208        self.enforce_retention()?;
209        self.retention_on_open_applied = true;
210        Ok(())
211    }
212
213    /// Captures both realtime and monotonic timestamps, similar to systemd's dual_timestamp_now().
214    ///
215    /// Returns (realtime_usec, monotonic_usec) where:
216    /// - realtime: microseconds since Unix epoch (CLOCK_REALTIME), monotonically increasing
217    /// - monotonic: microseconds since boot (CLOCK_MONOTONIC)
218    fn capture_dual_timestamp(
219        &mut self,
220        timestamp_override: Option<&EntryTimestamps>,
221    ) -> Result<(u64, u64)> {
222        let realtime = match timestamp_override.and_then(|ts| ts.entry_realtime_usec) {
223            Some(ts) => self.clock.observe(Microseconds::new(ts)).get(),
224            None => self.clock.now().get(),
225        };
226
227        let desired_monotonic = timestamp_override
228            .and_then(|ts| ts.entry_monotonic_usec)
229            .ok_or_else(|| {
230                WriterError::InvalidConfig("entry monotonic timestamp is required".to_string())
231            })?;
232
233        let monotonic = if desired_monotonic > self.last_monotonic_usec {
234            desired_monotonic
235        } else {
236            self.last_monotonic_usec.saturating_add(1)
237        };
238        self.last_monotonic_usec = monotonic;
239
240        Ok((realtime, monotonic))
241    }
242
243    fn require_entry_monotonic(timestamps: &EntryTimestamps) -> Result<()> {
244        if timestamps.entry_monotonic_usec.is_none() {
245            return Err(WriterError::InvalidConfig(
246                "entry monotonic timestamp is required".to_string(),
247            ));
248        }
249        Ok(())
250    }
251
252    /// Creates a new journal log.
253    pub fn new(path: &Path, config: Config) -> Result<Self> {
254        Self::new_inner(path, config, None, None)
255    }
256
257    pub fn new_with_lifecycle_observer(
258        path: &Path,
259        config: Config,
260        observer: Arc<dyn LogLifecycleObserver>,
261    ) -> Result<Self> {
262        Self::new_inner(path, config, Some(observer), None)
263    }
264
265    pub fn new_with_hooks(
266        path: &Path,
267        config: Config,
268        observer: Option<Arc<dyn LogLifecycleObserver>>,
269        artifact_sizer: Option<Arc<dyn LogArtifactSizer>>,
270    ) -> Result<Self> {
271        Self::new_inner(path, config, observer, artifact_sizer)
272    }
273
274    fn new_inner(
275        path: &Path,
276        config: Config,
277        lifecycle_observer: Option<Arc<dyn LogLifecycleObserver>>,
278        artifact_sizer: Option<Arc<dyn LogArtifactSizer>>,
279    ) -> Result<Self> {
280        let startup = build_startup_state(path, config)?;
281
282        let mut log = Log {
283            configured_dir: path.to_path_buf(),
284            chain: startup.chain,
285            config: startup.config,
286            active_file: startup.active_file,
287            rotation_state: startup.rotation_state,
288            boot_id: startup.boot_id,
289            seqnum_id: startup.seqnum_id,
290            current_seqnum: startup.current_seqnum,
291            clock: startup.clock,
292            last_monotonic_usec: startup.last_monotonic_usec,
293            lifecycle_observer,
294            artifact_sizer,
295            retention_on_open_applied: false,
296            boot_id_field: format!("_BOOT_ID={}", startup.boot_id.as_simple()).into_bytes(),
297            source_realtime_field: Vec::with_capacity(SOURCE_REALTIME_PREFIX.len() + 20),
298        };
299        if log.config.open_mode == LogOpenMode::Eager && log.active_file.is_none() {
300            let realtime = log.peek_entry_realtime(&EntryTimestamps::default());
301            log.rotate(realtime, LogLifecycleReason::EagerOpen)?;
302            log.retention_on_open_applied = true;
303        }
304        log.apply_retention_on_open()?;
305        Ok(log)
306    }
307
308    pub fn with_lifecycle_observer(mut self, observer: Arc<dyn LogLifecycleObserver>) -> Self {
309        self.lifecycle_observer = Some(observer);
310        self
311    }
312
313    pub fn with_artifact_sizer(mut self, sizer: Arc<dyn LogArtifactSizer>) -> Self {
314        self.artifact_sizer = Some(sizer);
315        self
316    }
317
318    /// Writes a journal entry.
319    ///
320    /// This compatibility method always returns an error under the strict
321    /// writer contract. Use [`Log::write_entry_with_timestamps`] and provide
322    /// an explicit entry monotonic timestamp.
323    #[deprecated(
324        since = "0.7.2",
325        note = "use write_entry_with_timestamps and provide an explicit entry monotonic timestamp"
326    )]
327    pub fn write_entry(
328        &mut self,
329        items: &[&[u8]],
330        source_realtime_usec: Option<u64>,
331    ) -> Result<()> {
332        self.write_entry_with_timestamps(
333            items,
334            EntryTimestamps {
335                source_realtime_usec,
336                ..EntryTimestamps::default()
337            },
338        )
339    }
340
341    /// Writes a journal entry with optional source and entry timestamp overrides.
342    ///
343    /// Overrides are safe by construction:
344    /// - entry realtime is clamped to strict monotonic progression (`last + 1us` floor)
345    /// - entry monotonic is also clamped to strict monotonic progression (`last + 1us` floor)
346    pub fn write_entry_with_timestamps(
347        &mut self,
348        items: &[&[u8]],
349        timestamps: EntryTimestamps,
350    ) -> Result<()> {
351        if items.is_empty() {
352            return Err(WriterError::EmptyEntry);
353        }
354        Self::require_entry_monotonic(&timestamps)?;
355
356        let entry_realtime = self.peek_entry_realtime(&timestamps);
357        self.prepare_append_for_realtime(entry_realtime)?;
358        let filtered_items = self.raw_items_for_policy(items)?;
359        let write_items = filtered_items.as_deref().unwrap_or(items);
360
361        let (realtime, monotonic) = self.capture_dual_timestamp(Some(&timestamps))?;
362        self.write_raw_entry_fields(
363            write_items,
364            timestamps.source_realtime_usec,
365            realtime,
366            monotonic,
367            self.low_level_entry_options(EntryWriteOptions::default()),
368        )?;
369
370        let active_file = self.active_file.as_ref().unwrap();
371        self.rotation_state.update(&active_file.writer);
372        self.current_seqnum += 1;
373
374        Ok(())
375    }
376
377    /// Writes a journal entry from structured field names and binary-safe values.
378    ///
379    /// This is the preferred path when the producer already has split field
380    /// names and values. If `source_realtime_usec` is provided, a
381    /// `_SOURCE_REALTIME_TIMESTAMP` field is added.
382    ///
383    /// This compatibility method always returns an error under the strict
384    /// writer contract. Use [`Log::write_fields_with_timestamps`] and provide
385    /// an explicit entry monotonic timestamp.
386    #[deprecated(
387        since = "0.7.2",
388        note = "use write_fields_with_timestamps and provide an explicit entry monotonic timestamp"
389    )]
390    pub fn write_fields(
391        &mut self,
392        fields: &[StructuredField<'_>],
393        source_realtime_usec: Option<u64>,
394    ) -> Result<()> {
395        self.write_fields_with_timestamps(
396            fields,
397            EntryTimestamps {
398                source_realtime_usec,
399                ..EntryTimestamps::default()
400            },
401        )
402    }
403
404    /// Writes structured fields with optional source and entry timestamp overrides.
405    ///
406    /// Entry monotonic timestamp is required. Entry realtime and monotonic
407    /// overrides use the same clamping rules as
408    /// [`Log::write_entry_with_timestamps`].
409    pub fn write_fields_with_timestamps(
410        &mut self,
411        fields: &[StructuredField<'_>],
412        timestamps: EntryTimestamps,
413    ) -> Result<()> {
414        self.write_fields_with_options(fields, timestamps, EntryWriteOptions::default())
415    }
416
417    /// Writes structured fields with explicit low-level entry write options.
418    ///
419    /// Use this only when the caller can satisfy any invariants required by the
420    /// selected [`EntryWriteOptions`], especially no duplicate full `KEY=value`
421    /// payloads when `trusted_unique_payloads` is enabled.
422    pub fn write_fields_with_options(
423        &mut self,
424        fields: &[StructuredField<'_>],
425        timestamps: EntryTimestamps,
426        options: EntryWriteOptions,
427    ) -> Result<()> {
428        if fields.is_empty() {
429            return Err(WriterError::EmptyEntry);
430        }
431        Self::require_entry_monotonic(&timestamps)?;
432
433        let entry_realtime = self.peek_entry_realtime(&timestamps);
434        self.prepare_append_for_realtime(entry_realtime)?;
435        let filtered_fields = self.structured_fields_for_policy(fields)?;
436        let write_fields = filtered_fields.as_deref().unwrap_or(fields);
437
438        let (realtime, monotonic) = self.capture_dual_timestamp(Some(&timestamps))?;
439        self.write_structured_entry_fields(
440            write_fields,
441            timestamps.source_realtime_usec,
442            realtime,
443            monotonic,
444            self.low_level_entry_options(options),
445        )?;
446
447        let active_file = self.active_file.as_ref().unwrap();
448        self.rotation_state.update(&active_file.writer);
449        self.current_seqnum += 1;
450
451        Ok(())
452    }
453
454    fn write_raw_entry_fields(
455        &mut self,
456        items: &[&[u8]],
457        source_realtime_usec: Option<u64>,
458        realtime: u64,
459        monotonic: u64,
460        options: EntryWriteOptions,
461    ) -> Result<()> {
462        let source_field = if let Some(timestamp_usec) = source_realtime_usec {
463            self.prepare_source_realtime_field(timestamp_usec);
464            Some(self.source_realtime_field.as_slice())
465        } else {
466            None
467        };
468
469        let total_items = items.len() + 1 + usize::from(source_field.is_some());
470        if total_items <= STACK_ENTRY_REF_LIMIT {
471            let mut refs = [EntryField::raw(&[]); STACK_ENTRY_REF_LIMIT];
472            let mut len = 0usize;
473            refs[len] = EntryField::raw(self.boot_id_field.as_slice());
474            len += 1;
475            if let Some(source_field) = source_field {
476                refs[len] = EntryField::raw(source_field);
477                len += 1;
478            }
479            for item in items {
480                refs[len] = EntryField::raw(item);
481                len += 1;
482            }
483            self.active_file.as_mut().unwrap().write_entry_fields(
484                refs[..len].iter().copied(),
485                realtime,
486                monotonic,
487                options,
488            )?;
489        } else {
490            let mut refs = Vec::with_capacity(total_items);
491            refs.push(EntryField::raw(self.boot_id_field.as_slice()));
492            if let Some(source_field) = source_field {
493                refs.push(EntryField::raw(source_field));
494            }
495            refs.extend(items.iter().copied().map(EntryField::raw));
496            self.active_file.as_mut().unwrap().write_entry_fields(
497                refs.iter().copied(),
498                realtime,
499                monotonic,
500                options,
501            )?;
502        }
503
504        Ok(())
505    }
506
507    fn write_structured_entry_fields(
508        &mut self,
509        fields: &[StructuredField<'_>],
510        source_realtime_usec: Option<u64>,
511        realtime: u64,
512        monotonic: u64,
513        options: EntryWriteOptions,
514    ) -> Result<()> {
515        let source_field = if let Some(timestamp_usec) = source_realtime_usec {
516            self.prepare_source_realtime_field(timestamp_usec);
517            Some(self.source_realtime_field.as_slice())
518        } else {
519            None
520        };
521
522        let total_items = fields.len() + 1 + usize::from(source_field.is_some());
523        if total_items <= STACK_ENTRY_REF_LIMIT {
524            let mut refs = [EntryField::raw(&[]); STACK_ENTRY_REF_LIMIT];
525            let mut len = 0usize;
526            refs[len] = EntryField::raw(self.boot_id_field.as_slice());
527            len += 1;
528            if let Some(source_field) = source_field {
529                refs[len] = EntryField::raw(source_field);
530                len += 1;
531            }
532            for field in fields {
533                refs[len] = EntryField::Structured(*field);
534                len += 1;
535            }
536            self.active_file.as_mut().unwrap().write_entry_fields(
537                refs[..len].iter().copied(),
538                realtime,
539                monotonic,
540                options,
541            )?;
542        } else {
543            let mut refs = Vec::with_capacity(total_items);
544            refs.push(EntryField::raw(self.boot_id_field.as_slice()));
545            if let Some(source_field) = source_field {
546                refs.push(EntryField::raw(source_field));
547            }
548            refs.extend(fields.iter().copied().map(EntryField::Structured));
549            self.active_file.as_mut().unwrap().write_entry_fields(
550                refs.iter().copied(),
551                realtime,
552                monotonic,
553                options,
554            )?;
555        }
556
557        Ok(())
558    }
559
560    fn low_level_entry_options(&self, options: EntryWriteOptions) -> EntryWriteOptions {
561        options.field_name_policy(log_writer_field_name_policy(self.config.field_name_policy))
562    }
563
564    fn prepare_source_realtime_field(&mut self, timestamp_usec: u64) {
565        self.source_realtime_field.clear();
566        self.source_realtime_field
567            .extend_from_slice(SOURCE_REALTIME_PREFIX);
568        let mut buffer = ItoaBuffer::new();
569        self.source_realtime_field
570            .extend_from_slice(buffer.format(timestamp_usec).as_bytes());
571    }
572
573    /// Syncs all written data to disk, ensuring durability.
574    ///
575    /// This should be called after writing a batch of log entries to ensure
576    /// they are persisted to disk before acknowledging the request.
577    pub fn sync(&mut self) -> Result<()> {
578        if let Some(active_file) = &mut self.active_file {
579            active_file.journal_file.sync()?;
580        }
581        Ok(())
582    }
583
584    /// Archives and closes the active file.
585    ///
586    /// In strict systemd naming mode this renames `<source>.journal` to the
587    /// chain filename before retention, matching the explicit close behavior of
588    /// the other SDK implementations. `Drop` remains best-effort for callers
589    /// that do not explicitly close.
590    pub fn close(mut self) -> Result<()> {
591        use journal_core::file::JournalState;
592
593        let Some(mut active_file) = self.active_file.take() else {
594            return Ok(());
595        };
596
597        let n_entries = active_file.journal_file.journal_header_ref().n_entries;
598        if self.config.strict_systemd_naming && n_entries == 0 {
599            match std::fs::remove_file(active_file.repository_file.path()) {
600                Ok(()) => {}
601                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
602                Err(err) => return Err(err.into()),
603            }
604            self.chain.remove_tracked_file(&active_file.repository_file);
605            return Ok(());
606        }
607
608        self.chain.update_file_size(
609            &active_file.repository_file,
610            active_file.current_file_size(),
611        );
612        active_file.journal_file.journal_header_mut().state = JournalState::Archived as u8;
613        sync_archive_journal_file(self.config.sync_on_archive, &mut active_file.journal_file)?;
614
615        let protected_file = if self.config.strict_systemd_naming {
616            let header = active_file.journal_file.journal_header_ref();
617            self.chain.archive_file(
618                &active_file.repository_file,
619                uuid::Uuid::from_bytes(header.seqnum_id),
620                header.head_entry_seqnum,
621                header.head_entry_realtime,
622            )?
623        } else {
624            active_file.repository_file.clone()
625        };
626
627        self.apply_retention(Some(&protected_file))?;
628
629        Ok(())
630    }
631
632    pub fn active_file(&self) -> Option<&repository::File> {
633        self.active_file
634            .as_ref()
635            .map(|active_file| &active_file.repository_file)
636    }
637
638    pub fn active_path(&self) -> Option<&Path> {
639        self.active_file
640            .as_ref()
641            .map(|active_file| Path::new(active_file.repository_file.path()))
642    }
643
644    pub fn configured_directory(&self) -> &Path {
645        &self.configured_dir
646    }
647
648    pub fn journal_directory(&self) -> &Path {
649        &self.chain.path
650    }
651
652    pub fn machine_id(&self) -> uuid::Uuid {
653        self.chain.machine_id
654    }
655
656    pub fn boot_id(&self) -> uuid::Uuid {
657        self.boot_id
658    }
659
660    pub fn source(&self) -> &journal_registry::Source {
661        &self.chain.source
662    }
663
664    /// Applies the configured retention policy without requiring a rotation or
665    /// close. The current active file is counted in retention envelopes and is
666    /// protected from deletion.
667    pub fn enforce_retention(&mut self) -> Result<()> {
668        let protected_file = if let Some(active_file) = &self.active_file {
669            self.chain.update_file_size(
670                &active_file.repository_file,
671                active_file.current_file_size(),
672            );
673            Some(active_file.repository_file.clone())
674        } else {
675            None
676        };
677        self.apply_retention(protected_file.as_ref())
678    }
679
680    fn update_active_file_size(&mut self) {
681        if let Some(active_file) = &self.active_file {
682            self.chain.update_file_size(
683                &active_file.repository_file,
684                active_file.current_file_size(),
685            );
686        }
687    }
688
689    fn prepare_initial_rotation(&mut self) -> Result<()> {
690        self.update_active_file_size();
691        if self.active_file.is_none() && self.config.strict_systemd_naming {
692            self.chain.archive_existing_active_file()?;
693        }
694        Ok(())
695    }
696
697    fn archive_rotated_file(&mut self, old_file: &ActiveFile) -> Result<repository::File> {
698        if !self.config.strict_systemd_naming {
699            return Ok(old_file.repository_file.clone());
700        }
701        let old_header = old_file.journal_file.journal_header_ref();
702        self.chain.archive_file(
703            &old_file.repository_file,
704            uuid::Uuid::from_bytes(old_header.seqnum_id),
705            old_header.head_entry_seqnum,
706            old_header.head_entry_realtime,
707        )
708    }
709
710    fn rotate_existing_active_file(
711        &mut self,
712        mut old_file: ActiveFile,
713        max_file_size: Option<u64>,
714        head_realtime: u64,
715    ) -> Result<(ActiveFile, LogLifecycleEvent)> {
716        use journal_core::file::JournalState;
717
718        old_file.journal_file.journal_header_mut().state = JournalState::Archived as u8;
719        sync_archive_journal_file(self.config.sync_on_archive, &mut old_file.journal_file)?;
720        let archived = self.archive_rotated_file(&old_file)?;
721        let new_file = old_file.rotate(
722            &mut self.chain,
723            max_file_size,
724            head_realtime,
725            self.config.compression,
726            self.config.compression_threshold,
727            self.config.strict_systemd_naming,
728            self.config.live_publish_every_entries,
729            self.config.file_mode,
730        )?;
731        let active = new_file.repository_file.clone();
732        Ok((new_file, LogLifecycleEvent::Rotated { archived, active }))
733    }
734
735    fn create_initial_active_file(
736        &mut self,
737        max_file_size: Option<u64>,
738        head_realtime: u64,
739        reason: LogLifecycleReason,
740    ) -> Result<(ActiveFile, LogLifecycleEvent)> {
741        let new_file = ActiveFile::create(
742            &mut self.chain,
743            self.seqnum_id,
744            self.boot_id,
745            self.current_seqnum + 1,
746            max_file_size,
747            head_realtime,
748            self.config.compression,
749            self.config.compression_threshold,
750            self.config.compact,
751            self.config.strict_systemd_naming,
752            self.config.live_publish_every_entries,
753            self.config.file_mode,
754        )?;
755        let active = new_file.repository_file.clone();
756        Ok((new_file, LogLifecycleEvent::Created { active, reason }))
757    }
758
759    fn emit_lifecycle_event(&self, event: &LogLifecycleEvent) {
760        if let Some(observer) = &self.lifecycle_observer {
761            observer.on_event(event);
762        }
763    }
764
765    fn protected_active_file(&self) -> Option<repository::File> {
766        self.active_file
767            .as_ref()
768            .map(|active_file| active_file.repository_file.clone())
769    }
770
771    #[tracing::instrument(skip_all, fields(active_file))]
772    fn rotate(&mut self, head_realtime: u64, reason: LogLifecycleReason) -> Result<()> {
773        self.prepare_initial_rotation()?;
774        let max_file_size = self.config.rotation_policy.size_of_journal_file;
775        let (new_file, lifecycle_event) = if let Some(old_file) = self.active_file.take() {
776            self.rotate_existing_active_file(old_file, max_file_size, head_realtime)?
777        } else {
778            self.create_initial_active_file(max_file_size, head_realtime, reason)?
779        };
780
781        tracing::Span::current().record("new_file", new_file.repository_file.path());
782
783        self.active_file = Some(new_file);
784        self.rotation_state.reset();
785        self.update_active_file_size();
786        self.emit_lifecycle_event(&lifecycle_event);
787
788        // Retention runs after the post-rotation current file is known, so the
789        // tracked current file counts in the envelope and is never deleted.
790        let protected_file = self.protected_active_file();
791        self.apply_retention(protected_file.as_ref())?;
792
793        Ok(())
794    }
795
796    /// Writes a journal entry from a serializable value.
797    ///
798    /// This method serializes the value to JSON, flattens it, and writes it to the journal.
799    /// The flattened structure converts nested JSON into KEY=VALUE pairs suitable for journal entries.
800    ///
801    /// # Example
802    ///
803    /// ```no_run
804    /// use serde::Serialize;
805    /// use journal_log_writer::{Config, EntryTimestamps, Log, RotationPolicy, RetentionPolicy};
806    /// use journal_registry::Origin;
807    /// use std::path::Path;
808    ///
809    /// #[derive(Serialize)]
810    /// struct LogEntry {
811    ///     message: String,
812    ///     level: String,
813    ///     user: User,
814    /// }
815    ///
816    /// #[derive(Serialize)]
817    /// struct User {
818    ///     id: u64,
819    ///     name: String,
820    /// }
821    ///
822    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
823    /// let origin = Origin {
824    ///     machine_id: Some("00112233445566778899aabbccddeeff".parse()?),
825    ///     namespace: None,
826    ///     source: journal_registry::Source::System,
827    /// };
828    /// let config = Config::new(origin, RotationPolicy::default(), RetentionPolicy::default())
829    ///     .with_boot_id("ffeeddccbbaa99887766554433221100".parse()?);
830    /// let mut log = Log::new(Path::new("/tmp/test-journal"), config)?;
831    ///
832    /// let entry = LogEntry {
833    ///     message: "User logged in".to_string(),
834    ///     level: "INFO".to_string(),
835    ///     user: User {
836    ///         id: 42,
837    ///         name: "alice".to_string(),
838    ///     },
839    /// };
840    ///
841    /// // This will write fields like:
842    /// // MESSAGE=User logged in
843    /// // LEVEL=INFO
844    /// // USER_ID=42
845    /// // USER_NAME=alice
846    /// let timestamps = EntryTimestamps::default()
847    ///     .with_entry_realtime_usec(1_700_000_000_000_000)
848    ///     .with_entry_monotonic_usec(1);
849    /// log.write_structured_with_timestamps(&entry, timestamps)?;
850    /// # Ok(())
851    /// # }
852    /// ```
853    #[cfg(feature = "serde-api")]
854    #[deprecated(
855        since = "0.7.2",
856        note = "use write_structured_with_timestamps and provide an explicit entry monotonic timestamp"
857    )]
858    pub fn write_structured<T: serde::Serialize>(&mut self, value: &T) -> Result<()> {
859        self.write_structured_with_timestamps(value, EntryTimestamps::default())
860    }
861
862    /// Writes a journal entry from a serializable value with explicit timestamps.
863    #[cfg(feature = "serde-api")]
864    pub fn write_structured_with_timestamps<T: serde::Serialize>(
865        &mut self,
866        value: &T,
867        timestamps: EntryTimestamps,
868    ) -> Result<()> {
869        // Serialize to JSON value
870        let json_value = serde_json::to_value(value).map_err(|e| {
871            WriterError::Serialization(format!("failed to serialize to JSON: {}", e))
872        })?;
873
874        // Flatten the JSON structure - requires a JSON object (Map)
875        let flattened = if let serde_json::Value::Object(map) = json_value {
876            flatten_json_map(&map)
877        } else {
878            // If not an object, return error
879            return Err(WriterError::Serialization(
880                "value must be a JSON object, not a primitive or array".to_string(),
881            ));
882        };
883
884        // Convert to journal field format (KEY=VALUE)
885        let mut fields: Vec<Vec<u8>> = Vec::with_capacity(flattened.len());
886
887        for (key, value) in flattened.iter() {
888            // Convert key to uppercase and replace dots with underscores
889            // (journal convention)
890            let journal_key = key.to_uppercase().replace('.', "_");
891
892            // Format as KEY=VALUE
893            let field = match value {
894                serde_json::Value::String(s) => {
895                    format!("{}={}", journal_key, s)
896                }
897                serde_json::Value::Number(n) => {
898                    format!("{}={}", journal_key, n)
899                }
900                serde_json::Value::Bool(b) => {
901                    format!("{}={}", journal_key, if *b { "true" } else { "false" })
902                }
903                serde_json::Value::Null => {
904                    format!("{}=", journal_key)
905                }
906                // Arrays and objects should be flattened already, but just in case
907                _ => {
908                    format!("{}={}", journal_key, value)
909                }
910            };
911
912            fields.push(field.into_bytes());
913        }
914
915        // Convert Vec<Vec<u8>> to Vec<&[u8]> for write_entry
916        let field_refs: Vec<&[u8]> = fields.iter().map(|f| f.as_slice()).collect();
917
918        self.write_entry_with_timestamps(&field_refs, timestamps)
919    }
920}
921
922#[cfg(all(test, feature = "serde-api"))]
923mod serde_api_tests;
924
925impl Drop for Log {
926    fn drop(&mut self) {
927        use journal_core::file::JournalState;
928
929        if let Some(ref mut active_file) = self.active_file {
930            // Keep the active path stable on close so file-backed readers that
931            // already follow system.journal can finish. The next writer startup
932            // archives this stale active file before creating a fresh one.
933            active_file.journal_file.journal_header_mut().state = JournalState::Archived as u8;
934
935            // Best/Last-effort sync just to be on the cautious side.
936            let _ = sync_archive_journal_file(
937                self.config.sync_on_archive,
938                &mut active_file.journal_file,
939            );
940        }
941    }
942}
943
944#[cfg(test)]
945mod tests {
946    use super::*;
947    use journal_registry::Origin;
948    use std::sync::Mutex;
949    use tempfile::TempDir;
950
951    static ARCHIVE_SYNC_TEST_LOCK: Mutex<()> = Mutex::new(());
952
953    fn test_uuid(seed: u8) -> uuid::Uuid {
954        uuid::Uuid::from_bytes([seed; 16])
955    }
956
957    fn test_config() -> Config {
958        Config::new(
959            Origin {
960                machine_id: Some(test_uuid(1)),
961                namespace: None,
962                source: journal_registry::Source::System,
963            },
964            RotationPolicy::default().with_number_of_entries(1),
965            RetentionPolicy::default(),
966        )
967        .with_boot_id(test_uuid(2))
968    }
969
970    fn write_test_entry(log: &mut Log, message: &[u8], realtime: u64) {
971        log.write_entry_with_timestamps(
972            &[message],
973            EntryTimestamps::default()
974                .with_entry_realtime_usec(realtime)
975                .with_entry_monotonic_usec(realtime),
976        )
977        .expect("write entry");
978    }
979
980    fn journal_paths(dir: &TempDir) -> Vec<PathBuf> {
981        let journal_dir = dir.path().join(test_uuid(1).as_simple().to_string());
982        let mut paths: Vec<_> = std::fs::read_dir(journal_dir)
983            .expect("read journal dir")
984            .map(|entry| entry.expect("read entry").path())
985            .filter(|path| path.extension().is_some_and(|ext| ext == "journal"))
986            .collect();
987        paths.sort();
988        paths
989    }
990
991    fn create_online_chain_active(dir: &TempDir) -> PathBuf {
992        let mut log = Log::new(
993            dir.path(),
994            test_config()
995                .with_rotation_policy(RotationPolicy::default().with_number_of_entries(10)),
996        )
997        .expect("create log");
998        write_test_entry(&mut log, b"MESSAGE=stale-online", 1);
999        log.sync().expect("sync stale online active");
1000        let active_path = log.active_path().expect("active path").to_path_buf();
1001        let active_file = log.active_file.take().expect("active file");
1002        drop(active_file);
1003        drop(log);
1004        active_path
1005    }
1006
1007    #[test]
1008    fn default_rotation_syncs_archived_file_on_caller_path() {
1009        let _guard = ARCHIVE_SYNC_TEST_LOCK.lock().expect("lock sync test");
1010        ARCHIVE_SYNC_CALLS.store(0, Ordering::Relaxed);
1011        let dir = tempfile::tempdir().expect("create temp dir");
1012        let mut log = Log::new(dir.path(), test_config()).expect("create log");
1013
1014        write_test_entry(&mut log, b"MESSAGE=first", 1);
1015        write_test_entry(&mut log, b"MESSAGE=second", 2);
1016
1017        assert_eq!(
1018            ARCHIVE_SYNC_CALLS.load(Ordering::Relaxed),
1019            1,
1020            "default rotation should sync the outgoing archived file"
1021        );
1022    }
1023
1024    #[test]
1025    fn sync_on_archive_false_skips_archive_sync_and_keeps_files_readable() {
1026        let _guard = ARCHIVE_SYNC_TEST_LOCK.lock().expect("lock sync test");
1027        ARCHIVE_SYNC_CALLS.store(0, Ordering::Relaxed);
1028        let dir = tempfile::tempdir().expect("create temp dir");
1029        let mut log =
1030            Log::new(dir.path(), test_config().with_sync_on_archive(false)).expect("create log");
1031
1032        write_test_entry(&mut log, b"MESSAGE=first", 1);
1033        write_test_entry(&mut log, b"MESSAGE=second", 2);
1034        log.close().expect("close log");
1035
1036        assert_eq!(
1037            ARCHIVE_SYNC_CALLS.load(Ordering::Relaxed),
1038            0,
1039            "opt-out must not sync archived files on the caller path"
1040        );
1041
1042        let paths = journal_paths(&dir);
1043        assert_eq!(
1044            paths.len(),
1045            2,
1046            "rotation plus close should leave two journals"
1047        );
1048        for path in paths {
1049            let file = JournalFile::<journal_core::file::Mmap>::open_path(&path, 32 * 1024 * 1024)
1050                .expect("open archived journal");
1051            assert_eq!(file.journal_header_ref().n_entries, 1);
1052        }
1053    }
1054
1055    #[test]
1056    fn sync_on_archive_false_skips_drop_archive_sync() {
1057        let _guard = ARCHIVE_SYNC_TEST_LOCK.lock().expect("lock sync test");
1058        ARCHIVE_SYNC_CALLS.store(0, Ordering::Relaxed);
1059        let dir = tempfile::tempdir().expect("create temp dir");
1060
1061        {
1062            let mut log = Log::new(dir.path(), test_config().with_sync_on_archive(false))
1063                .expect("create log");
1064            write_test_entry(&mut log, b"MESSAGE=drop-opt-out", 1);
1065        }
1066
1067        assert_eq!(
1068            ARCHIVE_SYNC_CALLS.load(Ordering::Relaxed),
1069            0,
1070            "opt-out must skip best-effort Drop archive sync"
1071        );
1072    }
1073
1074    #[test]
1075    fn strict_startup_sync_on_archive_policy_applies_to_online_chain_active() {
1076        let _guard = ARCHIVE_SYNC_TEST_LOCK.lock().expect("lock sync test");
1077
1078        let default_dir = tempfile::tempdir().expect("create default temp dir");
1079        let default_active = create_online_chain_active(&default_dir);
1080        ARCHIVE_SYNC_CALLS.store(0, Ordering::Relaxed);
1081        let _default_log = Log::new(
1082            default_dir.path(),
1083            test_config()
1084                .with_rotation_policy(RotationPolicy::default().with_number_of_entries(10))
1085                .with_strict_systemd_naming(true),
1086        )
1087        .expect("open strict log");
1088        assert_eq!(
1089            ARCHIVE_SYNC_CALLS.load(Ordering::Relaxed),
1090            1,
1091            "default strict startup should sync the stale online chain file"
1092        );
1093        let file =
1094            JournalFile::<journal_core::file::Mmap>::open_path(&default_active, 32 * 1024 * 1024)
1095                .expect("open default startup archived journal");
1096        assert_eq!(
1097            file.journal_header_ref().state,
1098            journal_core::file::JournalState::Archived as u8
1099        );
1100
1101        let opt_out_dir = tempfile::tempdir().expect("create opt-out temp dir");
1102        let opt_out_active = create_online_chain_active(&opt_out_dir);
1103        ARCHIVE_SYNC_CALLS.store(0, Ordering::Relaxed);
1104        let _opt_out_log = Log::new(
1105            opt_out_dir.path(),
1106            test_config()
1107                .with_rotation_policy(RotationPolicy::default().with_number_of_entries(10))
1108                .with_strict_systemd_naming(true)
1109                .with_sync_on_archive(false),
1110        )
1111        .expect("open strict opt-out log");
1112        assert_eq!(
1113            ARCHIVE_SYNC_CALLS.load(Ordering::Relaxed),
1114            0,
1115            "opt-out strict startup must not sync the stale online chain file"
1116        );
1117        let file =
1118            JournalFile::<journal_core::file::Mmap>::open_path(&opt_out_active, 32 * 1024 * 1024)
1119                .expect("open opt-out startup archived journal");
1120        assert_eq!(
1121            file.journal_header_ref().state,
1122            journal_core::file::JournalState::Archived as u8
1123        );
1124    }
1125}
journal_log_writer/log/mod.rs

journal_log_writer/log/
mod.rs
