codex_state/
log_db.rs

1//! Tracing log export into the local SQLite log database.
2//!
3//! This module provides a `tracing_subscriber::Layer` that captures events,
4//! formats each one into a `LogEntry`, and sends entries to a bounded background
5//! queue. The background task inserts into the dedicated `logs` SQLite database
6//! in batches to keep logging overhead low.
7//!
8//! ## Usage
9//!
10//! ```no_run
11//! use codex_state::log_db;
12//! use tracing_subscriber::prelude::*;
13//!
14//! # async fn example(state_db: std::sync::Arc<codex_state::StateRuntime>) {
15//! let layer = log_db::start(state_db);
16//! let _ = tracing_subscriber::registry()
17//!     .with(layer)
18//!     .try_init();
19//! # }
20//! ```
21
22use std::future::Future;
23use std::sync::OnceLock;
24use std::time::Duration;
25use std::time::SystemTime;
26use std::time::UNIX_EPOCH;
27
28use tokio::sync::mpsc;
29use tokio::sync::oneshot;
30use tracing::Event;
31use tracing::field::Field;
32use tracing::field::Visit;
33use tracing::level_filters::LevelFilter;
34use tracing::span::Attributes;
35use tracing::span::Id;
36use tracing::span::Record;
37use tracing_subscriber::Layer;
38use tracing_subscriber::field::RecordFields;
39use tracing_subscriber::filter::Targets;
40use tracing_subscriber::fmt::FormatFields;
41use tracing_subscriber::fmt::FormattedFields;
42use tracing_subscriber::fmt::format::DefaultFields;
43use tracing_subscriber::registry::LookupSpan;
44use uuid::Uuid;
45
46use crate::LogEntry;
47use crate::StateRuntime;
48
49const LOG_QUEUE_CAPACITY: usize = 512;
50const LOG_BATCH_SIZE: usize = 128;
51const LOG_FLUSH_INTERVAL: Duration = Duration::from_secs(2);
52
53pub fn default_filter() -> Targets {
54    Targets::new()
55        .with_default(LevelFilter::TRACE)
56        .with_target("hyper_util", LevelFilter::WARN)
57        .with_target("log", LevelFilter::OFF)
58        .with_target("codex_otel.log_only", LevelFilter::OFF)
59        .with_target("codex_otel.trace_safe", LevelFilter::OFF)
60        .with_target("rmcp::service", LevelFilter::INFO)
61        .with_target("codex_api::responses_websocket_timing", LevelFilter::OFF)
62}
63
64#[derive(Clone, Copy, Debug, Eq, PartialEq)]
65pub struct LogSinkQueueConfig {
66    pub queue_capacity: usize,
67    pub batch_size: usize,
68    pub flush_interval: Duration,
69}
70
71impl Default for LogSinkQueueConfig {
72    fn default() -> Self {
73        Self {
74            queue_capacity: LOG_QUEUE_CAPACITY,
75            batch_size: LOG_BATCH_SIZE,
76            flush_interval: LOG_FLUSH_INTERVAL,
77        }
78    }
79}
80
81impl LogSinkQueueConfig {
82    fn normalized(self) -> Self {
83        Self {
84            queue_capacity: self.queue_capacity.max(1),
85            batch_size: self.batch_size.max(1),
86            flush_interval: if self.flush_interval.is_zero() {
87                LOG_FLUSH_INTERVAL
88            } else {
89                self.flush_interval
90            },
91        }
92    }
93}
94
95/// A tracing log writer that can flush entries accepted by its queue.
96///
97/// Implementations should keep `Layer::on_event` non-blocking for ordinary log
98/// events. `flush` should wait for entries accepted before the flush command to
99/// be processed by the writer.
100pub trait LogWriter<S>: Layer<S>
101where
102    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
103{
104    fn flush(&self) -> impl Future<Output = ()> + Send + '_;
105}
106
107pub struct LogDbLayer {
108    sender: mpsc::Sender<LogDbCommand>,
109    process_uuid: String,
110}
111
112pub fn start(state_db: std::sync::Arc<StateRuntime>) -> LogDbLayer {
113    LogDbLayer::start(state_db)
114}
115
116impl Clone for LogDbLayer {
117    fn clone(&self) -> Self {
118        Self {
119            sender: self.sender.clone(),
120            process_uuid: self.process_uuid.clone(),
121        }
122    }
123}
124
125impl LogDbLayer {
126    pub fn start(state_db: std::sync::Arc<StateRuntime>) -> Self {
127        Self::start_with_config(state_db, LogSinkQueueConfig::default())
128    }
129
130    pub fn start_with_config(
131        state_db: std::sync::Arc<StateRuntime>,
132        config: LogSinkQueueConfig,
133    ) -> Self {
134        let config = config.normalized();
135        let (sender, receiver) = mpsc::channel(config.queue_capacity);
136        tokio::spawn(run_inserter(state_db, receiver, config));
137        Self {
138            sender,
139            process_uuid: current_process_log_uuid().to_string(),
140        }
141    }
142
143    pub async fn flush(&self) {
144        let (tx, rx) = oneshot::channel();
145        if self.sender.send(LogDbCommand::Flush(tx)).await.is_ok() {
146            let _ = rx.await;
147        }
148    }
149
150    fn try_send(&self, entry: LogEntry) {
151        let _ = self.sender.try_send(LogDbCommand::Entry(Box::new(entry)));
152    }
153}
154
155impl<S> Layer<S> for LogDbLayer
156where
157    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
158{
159    fn on_new_span(
160        &self,
161        attrs: &Attributes<'_>,
162        id: &Id,
163        ctx: tracing_subscriber::layer::Context<'_, S>,
164    ) {
165        let mut visitor = SpanFieldVisitor::default();
166        attrs.record(&mut visitor);
167
168        if let Some(span) = ctx.span(id) {
169            span.extensions_mut().insert(SpanLogContext {
170                name: span.metadata().name().to_string(),
171                formatted_fields: format_fields(attrs),
172                thread_id: visitor.thread_id,
173            });
174        }
175    }
176
177    fn on_record(
178        &self,
179        id: &Id,
180        values: &Record<'_>,
181        ctx: tracing_subscriber::layer::Context<'_, S>,
182    ) {
183        let mut visitor = SpanFieldVisitor::default();
184        values.record(&mut visitor);
185
186        if let Some(span) = ctx.span(id) {
187            let mut extensions = span.extensions_mut();
188            if let Some(log_context) = extensions.get_mut::<SpanLogContext>() {
189                if let Some(thread_id) = visitor.thread_id {
190                    log_context.thread_id = Some(thread_id);
191                }
192                append_fields(&mut log_context.formatted_fields, values);
193            } else {
194                extensions.insert(SpanLogContext {
195                    name: span.metadata().name().to_string(),
196                    formatted_fields: format_fields(values),
197                    thread_id: visitor.thread_id,
198                });
199            }
200        }
201    }
202
203    fn on_event(&self, event: &Event<'_>, ctx: tracing_subscriber::layer::Context<'_, S>) {
204        let metadata = event.metadata();
205        // `tracing-log` checks filters with the original log target before
206        // dispatching an event whose tracing target is `log`, so the outer
207        // target filter cannot reliably reject these bridged events.
208        if metadata.target() == "log" {
209            return;
210        }
211
212        // The SDK emits DEBUG timer meta-events every second per process; these
213        // were over 30% of retained logs in measured high-fanout Codex environments.
214        if metadata.target() == "opentelemetry_sdk"
215            && matches!(
216                *metadata.level(),
217                tracing::Level::TRACE | tracing::Level::DEBUG
218            )
219        {
220            return;
221        }
222
223        let mut visitor = MessageVisitor::default();
224        event.record(&mut visitor);
225        let thread_id = visitor
226            .thread_id
227            .clone()
228            .or_else(|| event_thread_id(event, &ctx));
229        let feedback_log_body = format_feedback_log_body(event, &ctx);
230
231        let now = SystemTime::now()
232            .duration_since(UNIX_EPOCH)
233            .unwrap_or_else(|_| Duration::from_secs(0));
234        let entry = LogEntry {
235            ts: now.as_secs() as i64,
236            ts_nanos: now.subsec_nanos() as i64,
237            level: metadata.level().as_str().to_string(),
238            target: metadata.target().to_string(),
239            message: visitor.message,
240            feedback_log_body: Some(feedback_log_body),
241            thread_id,
242            process_uuid: Some(self.process_uuid.clone()),
243            module_path: metadata.module_path().map(ToString::to_string),
244            file: metadata.file().map(ToString::to_string),
245            line: metadata.line().map(|line| line as i64),
246        };
247
248        self.try_send(entry);
249    }
250}
251
252impl<S> LogWriter<S> for LogDbLayer
253where
254    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
255{
256    fn flush(&self) -> impl Future<Output = ()> + Send + '_ {
257        LogDbLayer::flush(self)
258    }
259}
260
261enum LogDbCommand {
262    Entry(Box<LogEntry>),
263    Flush(oneshot::Sender<()>),
264}
265
266#[derive(Debug)]
267struct SpanLogContext {
268    name: String,
269    formatted_fields: String,
270    thread_id: Option<String>,
271}
272
273#[derive(Default)]
274struct SpanFieldVisitor {
275    thread_id: Option<String>,
276}
277
278impl SpanFieldVisitor {
279    fn record_field(&mut self, field: &Field, value: String) {
280        if field.name() == "thread_id" && self.thread_id.is_none() {
281            self.thread_id = Some(value);
282        }
283    }
284}
285
286impl Visit for SpanFieldVisitor {
287    fn record_i64(&mut self, field: &Field, value: i64) {
288        self.record_field(field, value.to_string());
289    }
290
291    fn record_u64(&mut self, field: &Field, value: u64) {
292        self.record_field(field, value.to_string());
293    }
294
295    fn record_bool(&mut self, field: &Field, value: bool) {
296        self.record_field(field, value.to_string());
297    }
298
299    fn record_f64(&mut self, field: &Field, value: f64) {
300        self.record_field(field, value.to_string());
301    }
302
303    fn record_str(&mut self, field: &Field, value: &str) {
304        self.record_field(field, value.to_string());
305    }
306
307    fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) {
308        self.record_field(field, value.to_string());
309    }
310
311    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
312        self.record_field(field, format!("{value:?}"));
313    }
314}
315
316fn event_thread_id<S>(
317    event: &Event<'_>,
318    ctx: &tracing_subscriber::layer::Context<'_, S>,
319) -> Option<String>
320where
321    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
322{
323    let mut thread_id = None;
324    if let Some(scope) = ctx.event_scope(event) {
325        for span in scope.from_root() {
326            let extensions = span.extensions();
327            if let Some(log_context) = extensions.get::<SpanLogContext>()
328                && log_context.thread_id.is_some()
329            {
330                thread_id = log_context.thread_id.clone();
331            }
332        }
333    }
334    thread_id
335}
336
337fn format_feedback_log_body<S>(
338    event: &Event<'_>,
339    ctx: &tracing_subscriber::layer::Context<'_, S>,
340) -> String
341where
342    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
343{
344    let mut feedback_log_body = String::new();
345    if let Some(scope) = ctx.event_scope(event) {
346        for span in scope.from_root() {
347            let extensions = span.extensions();
348            if let Some(log_context) = extensions.get::<SpanLogContext>() {
349                feedback_log_body.push_str(&log_context.name);
350                if !log_context.formatted_fields.is_empty() {
351                    feedback_log_body.push('{');
352                    feedback_log_body.push_str(&log_context.formatted_fields);
353                    feedback_log_body.push('}');
354                }
355            } else {
356                feedback_log_body.push_str(span.metadata().name());
357            }
358            feedback_log_body.push(':');
359        }
360        if !feedback_log_body.is_empty() {
361            feedback_log_body.push(' ');
362        }
363    }
364    feedback_log_body.push_str(&format_fields(event));
365    feedback_log_body
366}
367
368fn format_fields<R>(fields: R) -> String
369where
370    R: RecordFields,
371{
372    let formatter = DefaultFields::default();
373    let mut formatted = FormattedFields::<DefaultFields>::new(String::new());
374    let _ = formatter.format_fields(formatted.as_writer(), fields);
375    formatted.fields
376}
377
378fn append_fields(fields: &mut String, values: &Record<'_>) {
379    let formatter = DefaultFields::default();
380    let mut formatted = FormattedFields::<DefaultFields>::new(std::mem::take(fields));
381    let _ = formatter.add_fields(&mut formatted, values);
382    *fields = formatted.fields;
383}
384
385fn current_process_log_uuid() -> &'static str {
386    static PROCESS_LOG_UUID: OnceLock<String> = OnceLock::new();
387    PROCESS_LOG_UUID.get_or_init(|| {
388        let pid = std::process::id();
389        let process_uuid = Uuid::new_v4();
390        format!("pid:{pid}:{process_uuid}")
391    })
392}
393
394async fn run_inserter(
395    state_db: std::sync::Arc<StateRuntime>,
396    mut receiver: mpsc::Receiver<LogDbCommand>,
397    config: LogSinkQueueConfig,
398) {
399    let mut buffer = Vec::with_capacity(config.batch_size);
400    let mut ticker = tokio::time::interval(config.flush_interval);
401    // Consume the immediate startup tick so entries flush after the interval.
402    ticker.tick().await;
403    loop {
404        tokio::select! {
405            maybe_command = receiver.recv() => {
406                match maybe_command {
407                    Some(LogDbCommand::Entry(entry)) => {
408                        buffer.push(*entry);
409                        if buffer.len() >= config.batch_size {
410                            flush(&state_db, &mut buffer).await;
411                        }
412                    }
413                    Some(LogDbCommand::Flush(reply)) => {
414                        flush(&state_db, &mut buffer).await;
415                        let _ = reply.send(());
416                    }
417                    None => {
418                        flush(&state_db, &mut buffer).await;
419                        break;
420                    }
421                }
422            }
423            _ = ticker.tick() => {
424                flush(&state_db, &mut buffer).await;
425            }
426        }
427    }
428}
429
430async fn flush(state_db: &StateRuntime, buffer: &mut Vec<LogEntry>) {
431    if buffer.is_empty() {
432        return;
433    }
434    let entries = buffer.split_off(0);
435    let _ = state_db.insert_logs(entries.as_slice()).await;
436}
437
438#[derive(Default)]
439struct MessageVisitor {
440    message: Option<String>,
441    thread_id: Option<String>,
442}
443
444impl MessageVisitor {
445    fn record_field(&mut self, field: &Field, value: String) {
446        if field.name() == "message" && self.message.is_none() {
447            self.message = Some(value.clone());
448        }
449        if field.name() == "thread_id" && self.thread_id.is_none() {
450            self.thread_id = Some(value);
451        }
452    }
453}
454
455impl Visit for MessageVisitor {
456    fn record_i64(&mut self, field: &Field, value: i64) {
457        self.record_field(field, value.to_string());
458    }
459
460    fn record_u64(&mut self, field: &Field, value: u64) {
461        self.record_field(field, value.to_string());
462    }
463
464    fn record_bool(&mut self, field: &Field, value: bool) {
465        self.record_field(field, value.to_string());
466    }
467
468    fn record_f64(&mut self, field: &Field, value: f64) {
469        self.record_field(field, value.to_string());
470    }
471
472    fn record_str(&mut self, field: &Field, value: &str) {
473        self.record_field(field, value.to_string());
474    }
475
476    fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) {
477        self.record_field(field, value.to_string());
478    }
479
480    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
481        self.record_field(field, format!("{value:?}"));
482    }
483}
484
485#[cfg(test)]
486#[path = "log_db_filter_tests.rs"]
487mod filter_tests;
488
489#[cfg(test)]
490mod tests {
491    use std::io;
492    use std::sync::Arc;
493    use std::sync::Mutex;
494
495    use pretty_assertions::assert_eq;
496    use tracing_subscriber::filter::Targets;
497    use tracing_subscriber::fmt::writer::MakeWriter;
498    use tracing_subscriber::layer::SubscriberExt;
499    use tracing_subscriber::util::SubscriberInitExt;
500
501    use super::*;
502
503    fn temp_codex_home() -> std::path::PathBuf {
504        std::env::temp_dir().join(format!("codex-state-log-db-{}", Uuid::new_v4()))
505    }
506
507    async fn wait_for_log_count(runtime: &StateRuntime, expected: usize) -> Vec<crate::LogRow> {
508        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
509        loop {
510            let rows = runtime
511                .query_logs(&crate::LogQuery::default())
512                .await
513                .expect("query logs");
514            if rows.len() == expected {
515                return rows;
516            }
517            assert!(
518                tokio::time::Instant::now() < deadline,
519                "timed out waiting for {expected} logs; saw {}",
520                rows.len()
521            );
522            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
523        }
524    }
525
526    fn test_entry(message: &str) -> LogEntry {
527        LogEntry {
528            ts: 1,
529            ts_nanos: 2,
530            level: "INFO".to_string(),
531            target: "test".to_string(),
532            message: Some(message.to_string()),
533            feedback_log_body: Some(message.to_string()),
534            thread_id: Some("thread-1".to_string()),
535            process_uuid: Some("process-1".to_string()),
536            module_path: Some("module".to_string()),
537            file: Some("file.rs".to_string()),
538            line: Some(7),
539        }
540    }
541
542    #[derive(Clone, Default)]
543    struct SharedWriter {
544        bytes: Arc<Mutex<Vec<u8>>>,
545    }
546
547    impl SharedWriter {
548        fn snapshot(&self) -> String {
549            String::from_utf8(self.bytes.lock().expect("writer mutex poisoned").clone())
550                .expect("valid utf-8")
551        }
552    }
553
554    struct SharedWriterGuard {
555        bytes: Arc<Mutex<Vec<u8>>>,
556    }
557
558    impl<'a> MakeWriter<'a> for SharedWriter {
559        type Writer = SharedWriterGuard;
560
561        fn make_writer(&'a self) -> Self::Writer {
562            SharedWriterGuard {
563                bytes: Arc::clone(&self.bytes),
564            }
565        }
566    }
567
568    impl io::Write for SharedWriterGuard {
569        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
570            self.bytes
571                .lock()
572                .expect("writer mutex poisoned")
573                .extend_from_slice(buf);
574            Ok(buf.len())
575        }
576
577        fn flush(&mut self) -> io::Result<()> {
578            Ok(())
579        }
580    }
581
582    #[tokio::test]
583    async fn sqlite_feedback_logs_match_feedback_formatter_shape() {
584        let codex_home = temp_codex_home();
585        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
586            .await
587            .expect("initialize runtime");
588        let writer = SharedWriter::default();
589        let layer = start(runtime.clone());
590
591        let subscriber = tracing_subscriber::registry()
592            .with(
593                tracing_subscriber::fmt::layer()
594                    .with_writer(writer.clone())
595                    .with_ansi(false)
596                    .with_target(false)
597                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
598            )
599            .with(
600                layer
601                    .clone()
602                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
603            );
604        let guard = subscriber.set_default();
605
606        tracing::trace!("threadless-before");
607        tracing::info_span!("feedback-thread", thread_id = "thread-1", turn = 1).in_scope(|| {
608            tracing::info!(foo = 2, "thread-scoped");
609        });
610        tracing::debug!("threadless-after");
611
612        layer.flush().await;
613        drop(guard);
614
615        let feedback_logs = writer.snapshot();
616        let without_timestamps = |logs: &str| {
617            logs.lines()
618                .map(|line| match line.split_once(' ') {
619                    Some((_, rest)) => rest,
620                    None => line,
621                })
622                .collect::<Vec<_>>()
623                .join("\n")
624        };
625        let sqlite_logs = String::from_utf8(
626            runtime
627                .query_feedback_logs("thread-1")
628                .await
629                .expect("query feedback logs"),
630        )
631        .expect("valid utf-8");
632        assert_eq!(
633            without_timestamps(&sqlite_logs),
634            without_timestamps(&feedback_logs)
635        );
636
637        let _ = tokio::fs::remove_dir_all(codex_home).await;
638    }
639
640    #[tokio::test]
641    async fn flush_persists_logs_for_query() {
642        let codex_home = temp_codex_home();
643        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
644            .await
645            .expect("initialize runtime");
646        let layer = start(runtime.clone());
647
648        let guard = tracing_subscriber::registry()
649            .with(
650                layer
651                    .clone()
652                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
653            )
654            .set_default();
655
656        tracing::info!("buffered-log");
657
658        layer.flush().await;
659        drop(guard);
660
661        let after_flush = runtime
662            .query_logs(&crate::LogQuery::default())
663            .await
664            .expect("query logs after flush");
665        assert_eq!(after_flush.len(), 1);
666        assert_eq!(after_flush[0].message.as_deref(), Some("buffered-log"));
667
668        let _ = tokio::fs::remove_dir_all(codex_home).await;
669    }
670
671    #[tokio::test]
672    async fn configured_batch_size_flushes_without_explicit_flush() {
673        let codex_home = temp_codex_home();
674        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
675            .await
676            .expect("initialize runtime");
677        let layer = LogDbLayer::start_with_config(
678            runtime.clone(),
679            LogSinkQueueConfig {
680                queue_capacity: 8,
681                batch_size: 2,
682                flush_interval: std::time::Duration::from_secs(60),
683            },
684        );
685
686        let guard = tracing_subscriber::registry()
687            .with(
688                layer
689                    .clone()
690                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
691            )
692            .set_default();
693
694        tracing::info!("first-batch-log");
695        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
696        assert_eq!(
697            runtime
698                .query_logs(&crate::LogQuery::default())
699                .await
700                .expect("query logs before batch fills")
701                .len(),
702            0
703        );
704
705        tracing::info!("second-batch-log");
706        let after_batch = wait_for_log_count(&runtime, /*expected*/ 2).await;
707        drop(guard);
708
709        assert_eq!(
710            after_batch
711                .iter()
712                .map(|row| row.message.as_deref())
713                .collect::<Vec<_>>(),
714            vec![Some("first-batch-log"), Some("second-batch-log")]
715        );
716
717        let _ = tokio::fs::remove_dir_all(codex_home).await;
718    }
719
720    #[tokio::test]
721    async fn configured_flush_interval_persists_buffered_logs() {
722        let codex_home = temp_codex_home();
723        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
724            .await
725            .expect("initialize runtime");
726        let layer = LogDbLayer::start_with_config(
727            runtime.clone(),
728            LogSinkQueueConfig {
729                queue_capacity: 8,
730                batch_size: 128,
731                flush_interval: std::time::Duration::from_millis(10),
732            },
733        );
734        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
735
736        let guard = tracing_subscriber::registry()
737            .with(
738                layer
739                    .clone()
740                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
741            )
742            .set_default();
743
744        tracing::info!("interval-log");
745        let after_interval = wait_for_log_count(&runtime, /*expected*/ 1).await;
746        drop(guard);
747
748        assert_eq!(after_interval[0].message.as_deref(), Some("interval-log"));
749
750        let _ = tokio::fs::remove_dir_all(codex_home).await;
751    }
752
753    #[tokio::test]
754    async fn event_queue_drops_new_entries_when_full() {
755        let (sender, mut receiver) = mpsc::channel(1);
756        let layer = LogDbLayer {
757            sender,
758            process_uuid: "process-1".to_string(),
759        };
760
761        layer.try_send(test_entry("first-queued-log"));
762        layer.try_send(test_entry("dropped-log"));
763
764        match receiver.try_recv().expect("first entry queued") {
765            LogDbCommand::Entry(entry) => {
766                assert_eq!(entry.message.as_deref(), Some("first-queued-log"));
767            }
768            LogDbCommand::Flush(_) => panic!("expected queued entry"),
769        }
770        assert!(receiver.try_recv().is_err());
771    }
772
773    #[tokio::test]
774    async fn flush_waits_for_queue_capacity_and_receiver_processing() {
775        let (sender, mut receiver) = mpsc::channel(1);
776        let layer = LogDbLayer {
777            sender,
778            process_uuid: "process-1".to_string(),
779        };
780
781        layer.try_send(test_entry("queued-before-flush"));
782        let mut flush_task = tokio::spawn({
783            let layer = layer.clone();
784            async move {
785                layer.flush().await;
786            }
787        });
788
789        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
790        assert!(!flush_task.is_finished());
791
792        match receiver.recv().await.expect("queued entry") {
793            LogDbCommand::Entry(entry) => {
794                assert_eq!(entry.message.as_deref(), Some("queued-before-flush"));
795            }
796            LogDbCommand::Flush(_) => panic!("expected queued entry"),
797        }
798
799        match receiver.recv().await.expect("flush command") {
800            LogDbCommand::Flush(reply) => {
801                assert!(!flush_task.is_finished());
802                let _ = reply.send(());
803            }
804            LogDbCommand::Entry(_) => panic!("expected flush command"),
805        }
806
807        tokio::time::timeout(std::time::Duration::from_secs(1), &mut flush_task)
808            .await
809            .expect("flush task completes")
810            .expect("flush task succeeds");
811    }
812}
codex_state/log_db.rs

codex_state/
log_db.rs
