Merge branch 'main' into manual-pipeline-trigger

Author: Fluzko

.github/workflows/audit.yml

@@ -2,6 +2,7 @@ name: Security audit
 on:
   push:
     paths:
+      - '.github/workflows/audit.yml'
       - '**/Cargo.toml'
       - '**/Cargo.lock'
   schedule:
@@ -10,6 +11,7 @@ on:
 permissions:
   checks: write
   contents: read
+  issues: write
 
 jobs:
   security_audit:

.gitignore

@@ -7,3 +7,4 @@ target/
 .agents/skills/
 .claude/
 .zed/
+.progress/

Cargo.lock

@@ -22,7 +22,7 @@ repository = "https://github.com/dial9-rs/dial9-tokio-telemetry"
 
 [workspace.dependencies]
 libc = "0.2"
-dial9-perf-self-profile = { version = "0.4.3", path = "perf-self-profile" }
+dial9-perf-self-profile = { version = "0.4.4", path = "perf-self-profile" }
 dial9-trace-format = { version = "0.4.1", path = "dial9-trace-format" }
 dial9-trace-format-derive = { version = "0.4.1", path = "dial9-trace-format-derive" }
 dial9-macro = { version = "0.3.7", path = "dial9-macro" }

Cargo.toml

@@ -16,12 +16,12 @@ struct MainArgs {
 const MISSING_CONFIG_HELP: &str = "missing required `config` argument, e.g.\n  \
                            #[dial9_tokio_telemetry::main(config = my_config_fn)]\n\
                            or with an inline closure:\n  \
-                           #[dial9_tokio_telemetry::main(config = || Dial9Config::builder().base_path(...).max_total_size(...).build().unwrap())]";
+                           #[dial9_tokio_telemetry::main(config = || Dial9Config::builder().on_disk_buffer(...).max_total_size(...).build().unwrap())]";
 
 const CONFIG_MUST_BE_ZERO_ARG_HELP: &str = "`config` must be a zero-argument function path or a zero-argument closure, e.g.\n  \
                            #[dial9_tokio_telemetry::main(config = my_config_fn)]\n\
                            or with an inline closure:\n  \
-                           #[dial9_tokio_telemetry::main(config = || Dial9Config::builder().base_path(...).max_total_size(...).build().unwrap())]";
+                           #[dial9_tokio_telemetry::main(config = || Dial9Config::builder().on_disk_buffer(...).max_total_size(...).build().unwrap())]";
 impl Parse for MainArgs {
     fn parse(input: ParseStream) -> syn::Result<Self> {
         if input.is_empty() {
@@ -118,20 +118,19 @@ fn expand_main(args: MainArgs, input: ItemFn) -> Result<TokenStream2, syn::Error
 /// * `config` — a zero-argument function path or a zero-argument closure
 ///   returning any value convertible into a `TracedRuntime`. In
 ///   practice that means one of:
-///     - `Dial9Config` from `Dial9Config::builder().build()` (strict):
-///       any builder validation or writer-I/O failure surfaces from
-///       `.build()` as a `Dial9ConfigBuilderError`; runtime construction
+///     - `Dial9Config` from `Dial9Config::builder().on_disk_buffer(..)..build()`
+///       or `..in_memory_buffer()..build()` (strict): a writer-I/O failure surfaces
+///       from `.build()` as a `Dial9ConfigBuilderError`; runtime construction
 ///       under the macro panics on tokio-builder or telemetry-core I/O.
-///     - `Dial9Config` from `Dial9Config::builder().build_or_disabled()`
-///       (lenient): the same `Dial9Config` type, but validation and
-///       writer-I/O failures are logged at `error!` and downgraded to a
-///       disabled config that still preserves your `with_tokio`
-///       configurators.
+///     - `Dial9Config` from `..build_or_disabled()` (lenient): the same
+///       `Dial9Config` type, but writer-I/O failures are logged at `error!`
+///       and downgraded to a disabled config that still preserves your
+///       `with_tokio` configurators.
 ///     - The deprecated positional `dial9_tokio_telemetry::config::Dial9Config`,
 ///       kept compatible via a bridge impl.
 ///
-///   Use `.enabled(false)` on the builder to run without telemetry
-///   while keeping your `with_tokio` configurators.
+///   Use `.enabled(false)` to run without telemetry while keeping your
+///   `with_tokio` configurators.
 ///
 /// # Examples
 ///
@@ -142,7 +141,7 @@ fn expand_main(args: MainArgs, input: ItemFn) -> Result<TokenStream2, syn::Error
 ///
 /// fn my_config() -> Dial9Config {
 ///     Dial9Config::builder()
-///         .base_path("/tmp/trace.bin")
+///         .on_disk_buffer("/tmp/trace.bin")
 ///         .max_file_size(1024 * 1024)
 ///         .max_total_size(16 * 1024 * 1024)
 ///         .build()
@@ -164,7 +163,7 @@ fn expand_main(args: MainArgs, input: ItemFn) -> Result<TokenStream2, syn::Error
 /// ```rust,ignore
 /// #[dial9_tokio_telemetry::main(config = || {
 ///     Dial9Config::builder()
-///         .base_path("/tmp/trace.bin")
+///         .on_disk_buffer("/tmp/trace.bin")
 ///         .max_file_size(1024 * 1024)
 ///         .max_total_size(16 * 1024 * 1024)
 ///         .build()
@@ -181,7 +180,7 @@ fn expand_main(args: MainArgs, input: ItemFn) -> Result<TokenStream2, syn::Error
 /// ```rust,ignore
 /// #[dial9_tokio_telemetry::main(config = || {
 ///     Dial9Config::builder()
-///         .base_path("/tmp/trace.bin")
+///         .on_disk_buffer("/tmp/trace.bin")
 ///         .max_file_size(1024 * 1024)
 ///         .max_total_size(16 * 1024 * 1024)
 ///         .build_or_disabled()
@@ -197,6 +196,7 @@ fn expand_main(args: MainArgs, input: ItemFn) -> Result<TokenStream2, syn::Error
 /// ```rust,ignore
 /// #[dial9_tokio_telemetry::main(config = || {
 ///     Dial9Config::builder()
+///         .on_disk_buffer("/tmp/trace.bin")
 ///         .enabled(false)
 ///         .build()
 ///         .expect("config build failed")
@@ -205,6 +205,22 @@ fn expand_main(args: MainArgs, input: ItemFn) -> Result<TokenStream2, syn::Error
 ///     /* ... */
 /// }
 /// ```
+///
+/// In-memory writer (no telemetry on disk), select it with `.in_memory_buffer()`.
+///
+/// ```rust,ignore
+/// #[dial9_tokio_telemetry::main(config = || {
+///     Dial9Config::builder()
+///         .in_memory_buffer()
+///         .max_total_size(16 * 1024 * 1024)
+///         .with_runtime(|r| r.with_custom_pipeline(|p| p.pipe(MyUploader)))
+///         .build()
+///         .expect("config build failed")
+/// })]
+/// async fn main() {
+///     /* ... */
+/// }
+/// ```
 #[proc_macro_attribute]
 pub fn main(attr: TokenStream, item: TokenStream) -> TokenStream {
     let args = parse_macro_input!(attr as MainArgs);

dial9-macro/src/lib.rs

@@ -167,6 +167,10 @@ name = "pipeline_heap_bench"
 harness = false
 required-features = ["cpu-profiling"]
 
+[[example]]
+name = "block_in_place_workload"
+required-features = ["cpu-profiling"]
+
 [[example]]
 name = "blocking_pool_tracking"
 required-features = ["cpu-profiling"]
@@ -191,10 +195,6 @@ required-features = ["analysis"]
 name = "trace_to_jsonl"
 required-features = ["analysis"]
 
-[[example]]
-name = "trace_to_fat_jsonl"
-required-features = ["analysis"]
-
 [[example]]
 name = "kernel_sched_events"
 required-features = ["cpu-profiling", "analysis"]

dial9-tokio-telemetry/Cargo.toml

@@ -41,7 +41,7 @@ use dial9_tokio_telemetry::{main, Dial9Config, telemetry::TelemetryHandle};
 
 fn my_config() -> Dial9Config {
     Dial9Config::builder()
-        .base_path("/tmp/my_traces/trace.bin")
+        .on_disk_buffer("/tmp/my_traces/trace.bin")
         .max_total_size(5 * 1024 * 1024)   // keep at most 5 MiB on disk
         .max_file_size(1024 * 1024)     // optional: defaults to min(100 MiB, max_total_size / 4)
         .rotation_period(std::time::Duration::from_secs(300)) // optional: rotate every 5 min (default: 60 s)
@@ -167,7 +167,7 @@ fn my_config() -> Dial9Config {
         .build();
 
     Dial9Config::builder()
-        .base_path("/tmp/my_traces/trace.bin")
+        .on_disk_buffer("/tmp/my_traces/trace.bin")
         .max_file_size(100 * 1024 * 1024)
         .max_total_size(500 * 1024 * 1024)
         .with_tokio(|t| { t.worker_threads(4); })
@@ -440,6 +440,37 @@ record_event(
 # }
 ```
 
+### Custom event callbacks
+
+You can also register a callback that runs from dial9's flush thread and emits
+custom events. This is useful for draining application-owned queues or taking
+periodic snapshots without passing a [`TelemetryHandle`] through your code:
+
+```rust,no_run
+use dial9_trace_format::TraceEvent;
+use dial9_tokio_telemetry::telemetry::{CustomEventsConfig, TracedRuntime};
+
+#[derive(TraceEvent)]
+struct CacheEvent {
+    #[traceevent(timestamp)]
+    timestamp_ns: u64,
+    entries: u64,
+}
+
+let (_runtime, _guard) = TracedRuntime::builder()
+    .with_custom_events(CustomEventsConfig::default(), move |ctx| {
+        while let Ok(event) = rx.try_recv() {
+            ctx.record_event(event);
+        }
+    })
+    .build_and_start_with_writer(builder, writer)?;
+```
+
+`CustomEventsConfig::default()` runs the callback every flush cycle
+while telemetry is enabled, which fits drain-style callbacks. For polling-style
+callbacks, configure `minimum_interval(...)` to limit how often dial9 invokes
+the callback.
+
 ### Custom Runtime Hooks
 
 dial9 installs callbacks on all 8 Tokio runtime hooks to collect telemetry. If you need to run your own logic alongside dial9's instrumentation, use `with_tokio_hooks`:
@@ -498,7 +529,8 @@ fn my_config() -> Dial9Config {
         .build();
 
     Dial9Config::builder()
-        // ...
+        .on_disk_buffer("/tmp/dial9/trace.bin")
+        .max_total_size(1 << 30)
         .with_runtime(|r| {
             r.with_task_tracking(true)
              .with_s3_uploader(s3_config)