appcore_ops/
observation_file.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: observation_file.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/23 23:50:45 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 11:51:10 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Bounded asynchronous JSONL drain for local production operations.
12
13use crate::{ObservationEvent, ObservationSink, SharedObservationEvent};
14use parking_lot::Mutex;
15use serde::Serialize;
16use std::fs::{self, File, OpenOptions};
17use std::io::{BufRead, BufReader, Write};
18use std::path::{Path, PathBuf};
19use std::sync::atomic::{AtomicU64, Ordering};
20use std::sync::mpsc::{self, Receiver, SyncSender, TrySendError};
21use std::sync::Arc;
22use std::thread::JoinHandle;
23use std::time::Duration;
24
25const OBSERVATION_THREAD_STACK_BYTES: usize = 512 * 1024;
26
27/// Stable marker written as the first line of every observation JSONL file.
28pub const OBSERVATION_FILE_FORMAT_V1: &str = "# appcore-observations-v1";
29/// Maximum queued observation bytes, including the event currently being written.
30pub const MAX_FILE_OBSERVATION_QUEUE_BYTES: u64 = 8 * 1024 * 1024;
31/// Maximum item capacity accepted by the file observation queue.
32pub const MAX_FILE_OBSERVATION_QUEUE_ITEMS: usize = 65_536;
33/// Default deadline shared by flush queue admission and acknowledgement.
34pub const FILE_OBSERVATION_FLUSH_TIMEOUT: Duration = Duration::from_secs(30);
35const MAX_FILE_OBSERVATION_RECORD_BYTES: u64 = 256 * 1024;
36
37/// Local observation drain limits and retention policy.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct FileObservationSinkConfig {
40    /// Active JSONL file path.
41    pub path: PathBuf,
42    /// Maximum active file size before rotation.
43    pub max_file_bytes: u64,
44    /// Number of rotated files retained.
45    pub retained_files: usize,
46    /// Maximum queued observations awaiting disk.
47    pub queue_capacity: usize,
48    /// Number of records written between `fsync` calls.
49    pub sync_every_records: usize,
50}
51
52impl FileObservationSinkConfig {
53    /// Creates a production-safe bounded configuration.
54    pub fn new(path: impl Into<PathBuf>) -> Self {
55        Self {
56            path: path.into(),
57            max_file_bytes: 16 * 1024 * 1024,
58            retained_files: 4,
59            queue_capacity: 4_096,
60            sync_every_records: 64,
61        }
62    }
63
64    fn validate(&self) -> std::io::Result<()> {
65        if self.max_file_bytes < 64 * 1024
66            || self.retained_files == 0
67            || self.queue_capacity == 0
68            || self.queue_capacity > MAX_FILE_OBSERVATION_QUEUE_ITEMS
69            || self.sync_every_records == 0
70        {
71            return Err(std::io::Error::new(
72                std::io::ErrorKind::InvalidInput,
73                "observation drain limits must be positive, max_file_bytes >= 64 KiB and queue_capacity <= 65536",
74            ));
75        }
76        Ok(())
77    }
78}
79
80/// Point-in-time counters for one file observation drain.
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub struct FileObservationSinkStats {
83    /// Records durably accepted by the worker.
84    pub written: u64,
85    /// Records discarded because the bounded queue was full.
86    pub dropped: u64,
87    /// Filesystem or serialization failures.
88    pub errors: u64,
89}
90
91/// Aggregate memory pressure for the file observation queue.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub struct FileObservationSinkPressure {
94    /// Bytes retained by queued and currently written observations.
95    pub queued_bytes: u64,
96    /// Hard aggregate byte ceiling.
97    pub max_queue_bytes: u64,
98    /// Highest retained byte count observed since startup.
99    pub peak_queued_bytes: u64,
100    /// Events rejected because the aggregate byte ceiling was full.
101    pub byte_rejections: u64,
102}
103
104enum DrainCommand {
105    Event(QueuedEvent),
106    Flush(mpsc::Sender<()>),
107}
108
109struct QueuedEvent {
110    event: SharedObservationEvent,
111    line_bytes: u64,
112    _reservation: QueueReservation,
113}
114
115struct QueueReservation {
116    budget: Arc<QueueBudget>,
117    bytes: u64,
118}
119
120impl Drop for QueueReservation {
121    fn drop(&mut self) {
122        self.budget.used.fetch_sub(self.bytes, Ordering::AcqRel);
123    }
124}
125
126struct QueueBudget {
127    max_bytes: u64,
128    used: AtomicU64,
129    peak: AtomicU64,
130    rejected: AtomicU64,
131}
132
133impl QueueBudget {
134    const fn new(max_bytes: u64) -> Self {
135        Self {
136            max_bytes,
137            used: AtomicU64::new(0),
138            peak: AtomicU64::new(0),
139            rejected: AtomicU64::new(0),
140        }
141    }
142
143    fn reserve(self: &Arc<Self>, bytes: u64) -> Option<QueueReservation> {
144        let mut used = self.used.load(Ordering::Acquire);
145        loop {
146            let Some(next) = used.checked_add(bytes) else {
147                self.rejected.fetch_add(1, Ordering::Relaxed);
148                return None;
149            };
150            if next > self.max_bytes {
151                self.rejected.fetch_add(1, Ordering::Relaxed);
152                return None;
153            }
154            match self
155                .used
156                .compare_exchange_weak(used, next, Ordering::AcqRel, Ordering::Acquire)
157            {
158                Ok(_) => {
159                    self.peak.fetch_max(next, Ordering::Relaxed);
160                    return Some(QueueReservation {
161                        budget: Arc::clone(self),
162                        bytes,
163                    });
164                }
165                Err(current) => used = current,
166            }
167        }
168    }
169
170    fn pressure(&self) -> FileObservationSinkPressure {
171        FileObservationSinkPressure {
172            queued_bytes: self.used.load(Ordering::Acquire),
173            max_queue_bytes: self.max_bytes,
174            peak_queued_bytes: self.peak.load(Ordering::Relaxed),
175            byte_rejections: self.rejected.load(Ordering::Relaxed),
176        }
177    }
178}
179
180struct FileObservationSinkInner {
181    sender: Mutex<Option<SyncSender<DrainCommand>>>,
182    worker: Mutex<Option<JoinHandle<()>>>,
183    written: Arc<AtomicU64>,
184    dropped: AtomicU64,
185    errors: Arc<AtomicU64>,
186    queue_budget: Arc<QueueBudget>,
187    max_file_bytes: u64,
188}
189
190impl Drop for FileObservationSinkInner {
191    fn drop(&mut self) {
192        self.sender.get_mut().take();
193        if let Some(worker) = self.worker.get_mut().take() {
194            let _ = worker.join();
195        }
196    }
197}
198
199/// Cloneable non-blocking observation sink backed by a bounded worker queue.
200#[derive(Clone)]
201pub struct FileObservationSink {
202    inner: Arc<FileObservationSinkInner>,
203}
204
205impl std::fmt::Debug for FileObservationSink {
206    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
207        formatter
208            .debug_struct("FileObservationSink")
209            .field("stats", &self.stats())
210            .field("pressure", &self.pressure())
211            .finish()
212    }
213}
214
215impl FileObservationSink {
216    /// Creates the active file and starts the bounded writer worker.
217    pub fn new(config: FileObservationSinkConfig) -> std::io::Result<Self> {
218        config.validate()?;
219        initialize_file(&config.path)?;
220        let (sender, receiver) = mpsc::sync_channel(config.queue_capacity);
221        let written = Arc::new(AtomicU64::new(0));
222        let errors = Arc::new(AtomicU64::new(0));
223        let worker_written = Arc::clone(&written);
224        let worker_errors = Arc::clone(&errors);
225        let queue_budget = Arc::new(QueueBudget::new(MAX_FILE_OBSERVATION_QUEUE_BYTES));
226        let max_file_bytes = config.max_file_bytes;
227        let worker = std::thread::Builder::new()
228            .name("appcore-observation-drain".to_string())
229            .stack_size(OBSERVATION_THREAD_STACK_BYTES)
230            .spawn(move || run_worker(config, receiver, worker_written, worker_errors))?;
231        Ok(Self {
232            inner: Arc::new(FileObservationSinkInner {
233                sender: Mutex::new(Some(sender)),
234                worker: Mutex::new(Some(worker)),
235                written,
236                dropped: AtomicU64::new(0),
237                errors,
238                queue_budget,
239                max_file_bytes,
240            }),
241        })
242    }
243
244    /// Flushes all events accepted before this call.
245    pub fn flush(&self) -> std::io::Result<()> {
246        self.flush_timeout(FILE_OBSERVATION_FLUSH_TIMEOUT)
247    }
248
249    /// Flushes accepted events within one deadline covering queue admission
250    /// and worker acknowledgement.
251    pub fn flush_timeout(&self, timeout: Duration) -> std::io::Result<()> {
252        let deadline = crate::observation_flush::deadline(timeout)?;
253        let (acknowledge, receiver) = mpsc::channel();
254        let sender = self.inner.sender.lock().clone().ok_or_else(|| {
255            std::io::Error::new(std::io::ErrorKind::BrokenPipe, "observation drain stopped")
256        })?;
257        crate::observation_flush::enqueue(&sender, DrainCommand::Flush(acknowledge), deadline)?;
258        crate::observation_flush::wait(receiver, deadline)
259    }
260
261    /// Returns worker, backpressure and failure counters.
262    pub fn stats(&self) -> FileObservationSinkStats {
263        FileObservationSinkStats {
264            written: self.inner.written.load(Ordering::Relaxed),
265            dropped: self.inner.dropped.load(Ordering::Relaxed),
266            errors: self.inner.errors.load(Ordering::Relaxed),
267        }
268    }
269
270    /// Returns aggregate byte pressure for queued and currently written events.
271    pub fn pressure(&self) -> FileObservationSinkPressure {
272        self.inner.queue_budget.pressure()
273    }
274
275    fn enqueue_shared(&self, event: SharedObservationEvent) {
276        let line_bytes = match measure_event_line(self.inner.max_file_bytes, event.as_event()) {
277            Ok(bytes) => bytes,
278            Err(_) => {
279                self.inner.errors.fetch_add(1, Ordering::Relaxed);
280                return;
281            }
282        };
283        let Some(reservation) = self.inner.queue_budget.reserve(line_bytes) else {
284            self.inner.dropped.fetch_add(1, Ordering::Relaxed);
285            return;
286        };
287        let event = QueuedEvent {
288            event,
289            line_bytes,
290            _reservation: reservation,
291        };
292        let Some(sender) = self.inner.sender.lock().clone() else {
293            self.inner.dropped.fetch_add(1, Ordering::Relaxed);
294            return;
295        };
296        match sender.try_send(DrainCommand::Event(event)) {
297            Ok(()) => {}
298            Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
299                self.inner.dropped.fetch_add(1, Ordering::Relaxed);
300            }
301        }
302    }
303}
304
305impl ObservationSink for FileObservationSink {
306    fn emit(&self, event: ObservationEvent) {
307        self.enqueue_shared(SharedObservationEvent::new(event));
308    }
309
310    fn emit_shared(&self, event: &SharedObservationEvent) {
311        self.enqueue_shared(event.clone());
312    }
313}
314
315fn run_worker(
316    config: FileObservationSinkConfig,
317    receiver: Receiver<DrainCommand>,
318    written: Arc<AtomicU64>,
319    errors: Arc<AtomicU64>,
320) {
321    let mut unsynced = 0usize;
322    let mut file = open_append(&config.path).ok();
323    while let Ok(command) = receiver.recv() {
324        match command {
325            DrainCommand::Event(event) => {
326                let result =
327                    write_event(&config, &mut file, event.event.as_event(), event.line_bytes);
328                if result.is_ok() {
329                    written.fetch_add(1, Ordering::Relaxed);
330                    unsynced += 1;
331                } else {
332                    errors.fetch_add(1, Ordering::Relaxed);
333                }
334                if unsynced >= config.sync_every_records {
335                    sync_file(&mut file, &errors);
336                    unsynced = 0;
337                }
338            }
339            DrainCommand::Flush(acknowledge) => {
340                sync_file(&mut file, &errors);
341                unsynced = 0;
342                let _ = acknowledge.send(());
343            }
344        }
345    }
346    sync_file(&mut file, &errors);
347}
348
349fn write_event(
350    config: &FileObservationSinkConfig,
351    file: &mut Option<File>,
352    event: &ObservationEvent,
353    line_bytes: u64,
354) -> std::io::Result<()> {
355    let current_size = file
356        .as_ref()
357        .and_then(|file| file.metadata().ok())
358        .map(|metadata| metadata.len())
359        .unwrap_or(0);
360    if current_size
361        .checked_add(line_bytes)
362        .is_none_or(|bytes| bytes > config.max_file_bytes)
363    {
364        if let Some(active) = file.take() {
365            active.sync_all()?;
366        }
367        rotate_files(config)?;
368        *file = Some(open_append(&config.path)?);
369    }
370    if file.is_none() {
371        *file = Some(open_append(&config.path)?);
372    }
373    match file.as_mut() {
374        Some(file) => write_event_line(file, event),
375        None => Err(std::io::Error::other(
376            "observation file was not initialized",
377        )),
378    }
379}
380
381fn measure_event_line(max_file_bytes: u64, event: &ObservationEvent) -> std::io::Result<u64> {
382    let header_bytes = u64::try_from(OBSERVATION_FILE_FORMAT_V1.len())
383        .ok()
384        .and_then(|bytes| bytes.checked_add(1))
385        .ok_or_else(|| std::io::Error::other("observation size overflow"))?;
386    let file_json_limit = max_file_bytes
387        .checked_sub(header_bytes)
388        .and_then(|bytes| bytes.checked_sub(1))
389        .ok_or_else(record_too_large)?;
390    let record_json_limit = MAX_FILE_OBSERVATION_RECORD_BYTES
391        .checked_sub(1)
392        .ok_or_else(record_too_large)?;
393    let json_limit = file_json_limit.min(record_json_limit);
394    let mut counter = LimitedCounter::new(json_limit);
395    let result = serde_json::to_writer(&mut counter, &VersionedObservation::new(event));
396    if counter.exceeded {
397        return Err(record_too_large());
398    }
399    result.map_err(std::io::Error::other)?;
400    counter
401        .bytes
402        .checked_add(1)
403        .ok_or_else(|| std::io::Error::other("observation size overflow"))
404}
405
406fn write_event_line(writer: &mut impl Write, event: &ObservationEvent) -> std::io::Result<()> {
407    serde_json::to_writer(&mut *writer, &VersionedObservation::new(event))
408        .map_err(std::io::Error::other)?;
409    writer.write_all(b"\n")
410}
411
412fn record_too_large() -> std::io::Error {
413    std::io::Error::new(
414        std::io::ErrorKind::InvalidData,
415        "observation record exceeds file limit",
416    )
417}
418
419#[derive(Serialize)]
420struct VersionedObservation<'a> {
421    schema: &'static str,
422    event: &'a ObservationEvent,
423}
424
425impl<'a> VersionedObservation<'a> {
426    fn new(event: &'a ObservationEvent) -> Self {
427        Self {
428            schema: "appcore.observation.v1",
429            event,
430        }
431    }
432}
433
434struct LimitedCounter {
435    bytes: u64,
436    limit: u64,
437    exceeded: bool,
438}
439
440impl LimitedCounter {
441    const fn new(limit: u64) -> Self {
442        Self {
443            bytes: 0,
444            limit,
445            exceeded: false,
446        }
447    }
448}
449
450impl Write for LimitedCounter {
451    fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
452        let Some(bytes) = self.bytes.checked_add(buffer.len() as u64) else {
453            self.exceeded = true;
454            return Err(std::io::Error::other("observation size overflow"));
455        };
456        if bytes > self.limit {
457            self.exceeded = true;
458            return Err(record_too_large());
459        }
460        self.bytes = bytes;
461        Ok(buffer.len())
462    }
463
464    fn flush(&mut self) -> std::io::Result<()> {
465        Ok(())
466    }
467}
468
469fn initialize_file(path: &Path) -> std::io::Result<()> {
470    let parent = path.parent().unwrap_or_else(|| Path::new("."));
471    fs::create_dir_all(parent)?;
472    reject_symlink(path)?;
473    if !path.exists() {
474        let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
475        writeln!(file, "{OBSERVATION_FILE_FORMAT_V1}")?;
476        file.sync_all()?;
477        sync_parent(parent)?;
478        return Ok(());
479    }
480    let mut first = String::new();
481    BufReader::new(File::open(path)?).read_line(&mut first)?;
482    if first.trim_end() != OBSERVATION_FILE_FORMAT_V1 {
483        return Err(std::io::Error::new(
484            std::io::ErrorKind::InvalidData,
485            "unsupported observation file format",
486        ));
487    }
488    Ok(())
489}
490
491fn open_append(path: &Path) -> std::io::Result<File> {
492    initialize_file(path)?;
493    OpenOptions::new().append(true).read(true).open(path)
494}
495
496fn rotate_files(config: &FileObservationSinkConfig) -> std::io::Result<()> {
497    for index in (1..=config.retained_files).rev() {
498        let source = rotated_path(&config.path, index);
499        if index == config.retained_files {
500            remove_if_exists(&source)?;
501        } else if source.exists() {
502            fs::rename(&source, rotated_path(&config.path, index + 1))?;
503        }
504    }
505    if config.path.exists() {
506        fs::rename(&config.path, rotated_path(&config.path, 1))?;
507    }
508    initialize_file(&config.path)
509}
510
511fn rotated_path(path: &Path, index: usize) -> PathBuf {
512    let name = path
513        .file_name()
514        .and_then(|name| name.to_str())
515        .unwrap_or("observations.jsonl");
516    path.with_file_name(format!("{name}.{index}"))
517}
518
519fn reject_symlink(path: &Path) -> std::io::Result<()> {
520    match fs::symlink_metadata(path) {
521        Ok(metadata) if metadata.file_type().is_symlink() => Err(std::io::Error::new(
522            std::io::ErrorKind::InvalidInput,
523            "observation path must not be a symlink",
524        )),
525        Ok(metadata) if !metadata.is_file() => Err(std::io::Error::new(
526            std::io::ErrorKind::InvalidInput,
527            "observation path must be a regular file",
528        )),
529        Ok(_) => Ok(()),
530        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
531        Err(error) => Err(error),
532    }
533}
534
535fn sync_file(file: &mut Option<File>, errors: &AtomicU64) {
536    if file.as_ref().is_some_and(|file| file.sync_all().is_err()) {
537        errors.fetch_add(1, Ordering::Relaxed);
538    }
539}
540
541fn remove_if_exists(path: &Path) -> std::io::Result<()> {
542    match fs::remove_file(path) {
543        Ok(()) => Ok(()),
544        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
545        Err(error) => Err(error),
546    }
547}
548
549#[cfg(unix)]
550fn sync_parent(path: &Path) -> std::io::Result<()> {
551    File::open(path)?.sync_all()
552}
553
554#[cfg(not(unix))]
555fn sync_parent(_path: &Path) -> std::io::Result<()> {
556    Ok(())
557}
558
559#[cfg(test)]
560#[path = "observation_file_tests.rs"]
561mod tests;
appcore_ops/observation_file.rs

appcore_ops/
observation_file.rs
