Cleanups

Author: isbm

liblogjet/include/liblogjet.h

@@ -163,6 +163,7 @@ void lj_ingest_set_callback(lj_ingest_plugin *ctx, lj_record_callback cb, void *
 void lj_ingest_set_generic_callback(lj_ingest_plugin *ctx, lj_generic_record_callback cb, void *user);
 int lj_ingest_feed(lj_ingest_plugin *ctx, const uint8_t *data, size_t len);
 int lj_ingest_fetch(lj_ingest_plugin *ctx);
+const char *lj_ingest_last_error(lj_ingest_plugin *ctx);
 void lj_ingest_free(lj_ingest_plugin *ctx);
 
 #ifdef __cplusplus

logjetd/src/plugin.rs

@@ -449,7 +449,7 @@ fn push_unique_path(roots: &mut Vec<PathBuf>, path: PathBuf) {
     }
 }
 
-// ── C ABI types mirroring liblogjet.h ───────────────────────────────────────
+// C ABI types mirroring liblogjet.h
 
 // Legacy log-only signal mask (used when reserved[0] == 0 for old plugins).
 #[allow(dead_code)]
@@ -516,11 +516,12 @@ type SetCallbackFn = unsafe extern "C" fn(*mut LjIngestPlugin, RecordCallback, *
 type FeedFn = unsafe extern "C" fn(*mut LjIngestPlugin, *const u8, usize) -> c_int;
 type FetchFn = unsafe extern "C" fn(*mut LjIngestPlugin) -> c_int;
 type FreeFn = unsafe extern "C" fn(*mut LjIngestPlugin);
+type LastErrorFn = unsafe extern "C" fn(*mut LjIngestPlugin) -> *const c_char;
 type RecordCallback = unsafe extern "C" fn(*mut c_void, *const LjLogRecord);
 type GenericRecordCallback = unsafe extern "C" fn(*mut c_void, *const LjIngestRecordV1);
 type SetGenericCallbackFn = unsafe extern "C" fn(*mut LjIngestPlugin, GenericRecordCallback, *mut c_void);
 
-// ── Plugin handle ───────────────────────────────────────────────────────────
+// Plugin handle
 
 /// Resolved symbols from a loaded ingest plugin.
 struct PluginHandle {
@@ -534,6 +535,8 @@ struct PluginHandle {
     /// Multi-signal plugins export `lj_ingest_set_generic_callback`. If
     /// present, ljd calls it instead of `lj_ingest_set_callback`.
     set_generic_callback: Option<SetGenericCallbackFn>,
+    /// Optional error message retrieval (`lj_ingest_last_error`).
+    last_error: Option<LastErrorFn>,
     free: FreeFn,
 }
 
@@ -554,6 +557,8 @@ impl PluginHandle {
             let fetch: Option<FetchFn> = lib.get::<FetchFn>(b"lj_ingest_fetch\0").ok().map(|sym| *sym);
             let set_generic_callback: Option<SetGenericCallbackFn> =
                 lib.get::<SetGenericCallbackFn>(b"lj_ingest_set_generic_callback\0").ok().map(|sym| *sym);
+            let last_error: Option<LastErrorFn> =
+                lib.get::<LastErrorFn>(b"lj_ingest_last_error\0").ok().map(|sym| *sym);
             let free: libloading::Symbol<FreeFn> =
                 lib.get(b"lj_ingest_free\0").map_err(|err| io::Error::other(format!("symbol lj_ingest_free: {err}")))?;
 
@@ -563,6 +568,7 @@ impl PluginHandle {
                 feed: *feed,
                 fetch,
                 set_generic_callback,
+                last_error,
                 free: *free,
                 _lib: lib,
             })
@@ -575,7 +581,7 @@ impl PluginHandle {
     }
 }
 
-// ── Callback plumbing ───────────────────────────────────────────────────────
+// Callback plumbing
 
 /// Passed through the `void *user` pointer in the C callback.
 struct CallbackCtx {
@@ -777,7 +783,7 @@ pub(crate) fn build_otlp_payload(rec: OtlpRecord<'_>) -> Vec<u8> {
     request.encode_to_vec()
 }
 
-// ── Public entry point ──────────────────────────────────────────────────────
+// Public entry point
 
 /// Runs the plugin ingest loop: loads the .so, then either calls
 /// `lj_ingest_fetch` (active plugin) or binds TCP and feeds bytes (passive).
@@ -870,6 +876,15 @@ fn run_active_plugin(handle: &PluginHandle, spool: Arc<super::daemon::SharedSpoo
 
     let rc = unsafe { fetch(plugin_ctx) };
 
+    if rc != 0
+        && let Some(last_error_fn) = handle.last_error {
+            let msg = unsafe { last_error_fn(plugin_ctx) };
+            if !msg.is_null() {
+                let msg_str = unsafe { CStr::from_ptr(msg) }.to_string_lossy();
+                eprintln!("ljd plugin error: {msg_str}");
+            }
+        }
+
     unsafe { (handle.free)(plugin_ctx) };
     let _ = unsafe { Box::from_raw(ctx_ptr as *mut CallbackCtx) };
 

plugins/logcat-ingest/src/lib.rs

@@ -13,7 +13,7 @@ use std::ffi::{CString, c_char, c_int, c_void};
 use std::io::{self, BufRead, BufReader};
 use std::time::{SystemTime, UNIX_EPOCH};
 
-// ── C ABI types (must match liblogjet.h exactly) ────────────────────────────
+// C ABI types (must match liblogjet.h exactly)
 
 #[repr(C)]
 pub struct LjAttribute {
@@ -71,7 +71,7 @@ pub extern "C" fn lj_ingest_descriptor_v1() -> *const LjIngestDescriptorV1 {
     &LOGCAT_INGEST_DESCRIPTOR.0
 }
 
-// ── Severity constants ──────────────────────────────────────────────────────
+// Severity constants
 
 const LJ_SEVERITY_TRACE: i32 = 1;
 const LJ_SEVERITY_DEBUG: i32 = 5;
@@ -81,14 +81,14 @@ const LJ_SEVERITY_ERROR: i32 = 17;
 const LJ_SEVERITY_FATAL: i32 = 21;
 const LJ_ATTR_STRING: i32 = 0;
 
-// ── Plugin context ──────────────────────────────────────────────────────────
+// Plugin context
 
 pub struct LogcatPlugin {
     callback: Option<RecordCallback>,
     user: *mut c_void,
 }
 
-// ── Exported C ABI ──────────────────────────────────────────────────────────
+// Exported C ABI
 
 /// Creates a new logcat parsing context.
 #[unsafe(no_mangle)]
@@ -158,7 +158,7 @@ pub unsafe extern "C" fn lj_ingest_free(ctx: *mut LogcatPlugin) {
     let _ = unsafe { Box::from_raw(ctx) };
 }
 
-// ── Logcat parsing ──────────────────────────────────────────────────────────
+// Logcat parsing
 
 struct Parsed<'a> {
     severity: i32,
@@ -263,7 +263,7 @@ fn map_logcat_level(ch: u8) -> (i32, &'static str) {
     }
 }
 
-// ── Record emission ─────────────────────────────────────────────────────────
+// Record emission
 
 fn emit_record(ctx: &LogcatPlugin, line: &str) {
     let Some(cb) = ctx.callback else { return };

plugins/perfetto-ingest/src/lib.rs

@@ -15,7 +15,7 @@ mod trace_mapper;
 
 use std::ffi::{c_char, c_int, c_void};
 
-// ── C ABI types (must match liblogjet.h exactly) ────────────────────────────
+// C ABI types (must match liblogjet.h exactly)
 
 #[repr(C)]
 pub struct LjAttribute {
@@ -66,7 +66,7 @@ pub struct LjIngestDescriptorV1 {
     reserved: [u64; 8],
 }
 
-// ── Signal constants ────────────────────────────────────────────────────────
+// Signal constants
 
 const LJ_INGEST_SIGNAL_LOGS: u32 = 1 << 0;
 const LJ_INGEST_SIGNAL_METRICS: u32 = 1 << 1;
@@ -82,7 +82,7 @@ const LJ_INGEST_RECORD_TYPE_TRACES: u32 = 3;
 #[allow(dead_code)]
 const LJ_INGEST_RECORD_TYPE_EVENTS: u32 = 4;
 
-// ── Descriptor ──────────────────────────────────────────────────────────────
+// Descriptor
 
 struct IngestDescriptor(LjIngestDescriptorV1);
 
@@ -107,16 +107,17 @@ pub extern "C" fn lj_ingest_descriptor_v1() -> *const LjIngestDescriptorV1 {
     &PERFETTO_INGEST_DESCRIPTOR.0
 }
 
-// ── Plugin context ──────────────────────────────────────────────────────────
+// Plugin context
 
 pub struct PerfettoPlugin {
     pub(crate) legacy_callback: Option<RecordCallback>,
     pub(crate) legacy_user: *mut c_void,
     pub(crate) generic_callback: Option<GenericRecordCallback>,
     pub(crate) generic_user: *mut c_void,
+    last_error: Option<String>,
 }
 
-// ── Exported C ABI ──────────────────────────────────────────────────────────
+// Exported C ABI
 
 #[unsafe(no_mangle)]
 pub extern "C" fn lj_ingest_create() -> *mut PerfettoPlugin {
@@ -125,6 +126,7 @@ pub extern "C" fn lj_ingest_create() -> *mut PerfettoPlugin {
         legacy_user: std::ptr::null_mut(),
         generic_callback: None,
         generic_user: std::ptr::null_mut(),
+        last_error: None,
     }))
 }
 
@@ -164,6 +166,23 @@ pub unsafe extern "C" fn lj_ingest_feed(_ctx: *mut PerfettoPlugin, _data: *const
     0
 }
 
+/// Returns the last error message, or NULL if none.
+///
+/// # Safety
+///
+/// `ctx` must be a valid pointer from `lj_ingest_create`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn lj_ingest_last_error(ctx: *mut PerfettoPlugin) -> *const c_char {
+    if ctx.is_null() {
+        return std::ptr::null();
+    }
+    let ctx = unsafe { &*ctx };
+    match &ctx.last_error {
+        Some(msg) => msg.as_ptr().cast::<c_char>(),
+        None => std::ptr::null(),
+    }
+}
+
 /// Active source: reads a `.pftrace` file, invokes trace_processor, maps
 /// results to OTel, and streams records through the generic callback.
 ///
@@ -176,31 +195,32 @@ pub unsafe extern "C" fn lj_ingest_fetch(ctx: *mut PerfettoPlugin) -> c_int {
         eprintln!("perfetto-ingest: lj_ingest_fetch called with null context");
         return -1;
     }
-    let ctx = unsafe { &*ctx };
+    let ctx = unsafe { &mut *ctx };
 
     let trace_file = match std::env::var("LJD_PERFETTO_TRACE_FILE") {
         Ok(path) => std::path::PathBuf::from(path),
         Err(_) => {
-            eprintln!("perfetto-ingest: LJD_PERFETTO_TRACE_FILE is not set");
+            ctx.last_error = Some("LJD_PERFETTO_TRACE_FILE is not set".to_string());
             return -2;
         }
     };
 
     if !trace_file.is_file() {
-        eprintln!("perfetto-ingest: trace file not found: {}", trace_file.display());
+        ctx.last_error = Some(format!("trace file not found: {}", trace_file.display()));
         return -3;
     }
 
     match run_pipeline(ctx, &trace_file) {
         Ok(()) => 0,
         Err(err) => {
+            ctx.last_error = Some(err.to_string());
             eprintln!("perfetto-ingest: {err}");
             -4
         }
     }
 }
 
-fn run_pipeline(plugin: &PerfettoPlugin, trace_file: &std::path::Path) -> Result<(), String> {
+fn run_pipeline(plugin: &mut PerfettoPlugin, trace_file: &std::path::Path) -> Result<(), String> {
     let tp_path = perfetto_invoke::find_trace_processor()
         .map_err(|err| format!("trace_processor not found: {err}"))?;
 
@@ -270,7 +290,7 @@ pub unsafe extern "C" fn lj_ingest_free(ctx: *mut PerfettoPlugin) {
     let _ = unsafe { Box::from_raw(ctx) };
 }
 
-// ── Record emission helpers ─────────────────────────────────────────────────
+// Record emission helpers
 
 /// Calls the generic callback with a pre-encoded OTLP payload.
 ///

plugins/perfetto-ingest/src/sqlite_reader.rs

@@ -8,7 +8,7 @@
 
 use std::path::Path;
 
-// ── Typed models ────────────────────────────────────────────────────────────
+// Typed models
 
 #[derive(Debug, Clone)]
 pub struct PerfettoSlice {
@@ -69,7 +69,7 @@ pub struct PerfettoClockSnapshot {
     pub clock_value: i64,
 }
 
-// ── Database reader ──────────────────────────────────────────────────────────
+// Database reader
 
 pub struct PerfettoDb {
     pub(crate) conn: rusqlite::Connection,

plugins/perfetto-ingest/src/trace_mapper.rs

@@ -192,7 +192,7 @@ fn flush_batch(
     Ok(())
 }
 
-// ── ID generation ──────────────────────────────────────────────────────────
+// ID generation
 
 fn make_trace_id() -> [u8; 16] {
     let mut id = [0u8; 16];
@@ -211,7 +211,7 @@ fn make_span_id(slice_id: i64) -> [u8; 8] {
     id
 }
 
-// ── Attribute helpers ──────────────────────────────────────────────────────
+// Attribute helpers
 
 fn key_value(key: &str, value: AnyValue) -> KeyValue {
     KeyValue { key: key.to_string(), value: Some(value) }

plugins/perfetto-ingest/tests/unit/tests.rs

@@ -6,7 +6,7 @@ use prost::Message;
 /// Captured emitted record: (record_type, timestamp_unix_ns, payload).
 type EmittedRecord = (u32, u64, Vec<u8>);
 
-// ── sqlite_reader tests ─────────────────────────────────────────────────────
+// sqlite_reader tests
 
 #[test]
 fn sqlite_reader_reads_slices_ordered_by_ts() {
@@ -110,7 +110,7 @@ fn sqlite_reader_reads_clock_snapshots() {
     assert_eq!(snaps[1].clock_value, 1_700_000_000_000_010_000);
 }
 
-// ── metrics_reader tests ────────────────────────────────────────────────────
+// metrics_reader tests
 
 #[test]
 fn metrics_reader_parses_scalar_metric() {
@@ -154,7 +154,7 @@ fn metrics_reader_parses_nested_metric() {
     assert_eq!(metrics[0].children[0].scalar_value, Some(5.0));
 }
 
-// ── helpers ─────────────────────────────────────────────────────────────────
+// helpers
 
 fn temp_json(name: &str, content: &str) -> std::path::PathBuf {
     let dir = std::env::temp_dir();
@@ -199,7 +199,7 @@ fn test_db() -> super::sqlite_reader::PerfettoDb {
     super::sqlite_reader::PerfettoDb { conn }
 }
 
-// ── timestamp tests ─────────────────────────────────────────────────────────
+// timestamp tests
 
 fn make_snapshots(pairs: &[(i64, i64)]) -> Vec<crate::sqlite_reader::PerfettoClockSnapshot> {
     pairs.iter().map(|(ts, cv)| crate::sqlite_reader::PerfettoClockSnapshot { ts: *ts, clock_value: *cv }).collect()
@@ -267,7 +267,7 @@ fn timestamp_has_realtime() {
     assert!(conv.has_realtime());
 }
 
-// ── trace_mapper tests ──────────────────────────────────────────────────────
+// trace_mapper tests
 
 #[test]
 fn trace_mapper_produces_spans_from_slices() {
@@ -301,7 +301,7 @@ fn trace_mapper_fails_without_realtime_require() {
     assert!(result.is_err());
 }
 
-// ── metric_mapper tests ─────────────────────────────────────────────────────
+// metric_mapper tests
 
 #[test]
 fn metric_mapper_encodes_scalar_metrics() {
@@ -355,7 +355,7 @@ fn metric_mapper_flattens_nested_metrics() {
     assert!(names.contains(&"parent.child"));
 }
 
-// ── log_mapper tests ────────────────────────────────────────────────────────
+// log_mapper tests
 
 #[test]
 fn log_mapper_produces_summary_log() {
@@ -376,7 +376,7 @@ fn log_mapper_produces_summary_log() {
             }
 }
 
-// ── run_pipeline integration test ─────────────────────────────────────────────
+// run_pipeline integration test
 
 #[test]
 fn run_pipeline_integration_with_sqlite() {
@@ -395,7 +395,7 @@ fn run_pipeline_integration_with_sqlite() {
     let _ = std::fs::remove_file(&tmp);
 }
 
-// ── test helpers for mapper tests ───────────────────────────────────────────
+// test helpers for mapper tests
 
 fn run_trace_mapper(
     db: &super::sqlite_reader::PerfettoDb,
@@ -445,6 +445,7 @@ fn dummy_plugin(cb: super::GenericRecordCallback) -> super::PerfettoPlugin {
         legacy_user: std::ptr::null_mut(),
         generic_callback: Some(cb),
         generic_user: user_ptr,
+        last_error: None,
     }
 }