Breadway/bread
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add event causality tracking (id + caused_by chains)

Browse source 
Every BreadEvent now gets a unique id at construction (via
BreadEvent::new/with_timestamp, and the normalizer's struct-literal call
sites, which all now assign one explicitly). A new caused_by field is
populated only when a Lua bread.emit() call runs synchronously inside a
bread.on subscriber's handler: LuaEngine tracks the currently-dispatching
event's id (set/restored around each handle_event invocation, single-
threaded so a plain Mutex-guarded slot suffices) and bread.emit()'s Rust
binding reads it when constructing the outgoing event. This lets chains of
Lua modules that react to each other's events be reconstructed instead of
timestamp-guessed from a live `bread events` log.

Also:
- bread events --tree renders the caused_by chain as an indented,
  live-streaming-friendly tree instead of a flat line-per-event stream.
- API_VERSION bumped 1.4.0 -> 1.5.0 (additive-only change).
- Documentation.md's event envelope, bread.emit, and debugging-tips
  sections updated with Since: v1.5 markers.
- New end-to-end regression test spawning the real daemon with 3 chained
  Lua handlers (A emits X on trigger, B emits Y on X, C emits Z on Y) and
  asserting caused_by threads correctly through all three hops.
This commit is contained in:
Breadway 2026-08-04 17:55:31 +08:00
parent 96639516b1
commit 6ff1ee910b
10 changed files with 420 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

12
Cargo.lock generated
View file

@ -325,6 +325,7 @@ dependencies = [
"serde",
"serde_json",
"toml",
"uuid",
]
[[package]]
@ -2051,6 +2052,17 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "uuid"
version = "1.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
dependencies = [
"getrandom 0.4.3",
"js-sys",
"wasm-bindgen",
]
[[package]]
name = "valuable"
version = "0.1.1"

10
Documentation.md
View file

@ -152,6 +152,7 @@ installed_at = "2026-01-01T00:00:00Z"
## Debugging tips
- Run `bread events` to see live normalized events.
- Run `bread events --tree` *(Since: v1.5)* to render events as a causality tree instead of a flat stream — events that a Lua handler emitted via `bread.emit()` in reaction to another event are nested underneath it, following the `caused_by` chain (see [Dictionary: Event reference](#dictionary-event-reference)). Useful for untangling "why did this event fire" when several modules chain-react to each other.
- Run `bread state` to see full runtime state as JSON.
- Run `bread doctor` to check adapter and module health.
- Log event payloads with `bread.log(tostring(event.data))`.
@ -213,7 +214,7 @@ end, {
Unsubscribe an event handler or state watch by ID.
#### `bread.emit(event, data)`
Emit a custom event into the system pipeline. Useful for cross-module communication.
Emit a custom event into the system pipeline. Useful for cross-module communication. If called synchronously from inside a `bread.on` subscriber callback (i.e. in reaction to a matched event), the emitted event's `caused_by` *(Since: v1.5)* is set to the id of the event that triggered the callback, threading causality across chains of modules that react to each other — see [Dictionary: Event reference](#dictionary-event-reference).
#### `bread.wait(pattern, opts) -> event | nil`
Coroutine-only helper that suspends until a matching event arrives.
@ -855,10 +856,15 @@ Events are delivered as a `BreadEvent`:
"event": "bread.device.dock.connected",
"timestamp": 1710000000000,
"source": "Udev",
"data": {}
"data": {},
"id": "b3f2c9a0-4e6d-4b8a-9c1e-7a2f5d8e0c11",
"caused_by": null
}
```
- **`id`** *(Since: v1.5)* — a unique id assigned to this specific event instance at construction. Every `BreadEvent`, regardless of origin (adapter-normalized, IPC `emit`, Lua `bread.emit()`, or a daemon-internal send like `bread.system.startup`), gets one.
- **`caused_by`** *(Since: v1.5)* — the `id` of the event whose Lua subscriber handler emitted this event via `bread.emit()`, or `null` if this event did not originate from inside a running handler (adapter events, IPC `emit`, daemon-internal sends). This lets you reconstruct causality chains across modules that react to each other's events: if module A's handler for event X calls `bread.emit("Y", ...)`, then Y's `caused_by` is X's `id`. See `bread events --tree` below for a rendering of these chains.
### Pattern matching
| Pattern | Matches |

81
bread-cli/src/main.rs
View file

@ -6,6 +6,7 @@ use anyhow::Result;
use clap::{Parser, Subcommand};
use notify::{RecommendedWatcher, RecursiveMode, Watcher};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::env;
use std::io::{self, Write as IoWrite};
use std::path::{Path, PathBuf};
@ -54,6 +55,12 @@ enum Commands {
/// Replay events from the last N seconds
#[arg(long)]
since: Option<u64>,
/// Render events as a causality tree via `caused_by` instead of a
/// flat stream (events emitted from inside a `bread.emit()` call
/// made by another event's Lua handler are nested under it).
/// Overrides `--json` — tree rendering always uses the formatted view.
#[arg(long)]
tree: bool,
},
/// Manage installed Lua modules
Modules {
@ -158,8 +165,9 @@ async fn main() -> Result<()> {
json,
fields,
since,
tree,
} => {
stream_events(&socket, pattern, json, fields, since).await?;
stream_events(&socket, pattern, json, fields, since, tree).await?;
}
Commands::Modules { subcommand } => {
handle_modules_cmd(subcommand, &socket).await?;
@ -378,7 +386,13 @@ async fn stream_events(
raw_json: bool,
fields: Option<String>,
since: Option<u64>,
tree: bool,
) -> Result<()> {
// Tree rendering needs `id`/`caused_by` visible in a consistent shape,
// so it always uses the formatted view — a live-streaming-friendly
// indent-as-you-go tree rather than buffering the whole stream.
let mut causality = CausalityTracker::default();
if let Some(seconds) = since {
let replay = send_request(
socket,
@ -388,7 +402,9 @@ async fn stream_events(
.await?;
if let Some(list) = replay.as_array() {
for item in list {
if raw_json {
if tree {
causality.print(item);
} else if raw_json {
println!("{}", serde_json::to_string_pretty(item)?);
} else {
print_event(item, fields.as_deref());
@ -426,7 +442,9 @@ async fn stream_events(
while let Some(line) = lines.next_line().await? {
let value: Value = serde_json::from_str(&line)?;
if raw_json {
if tree {
causality.print(&value);
} else if raw_json {
println!("{}", serde_json::to_string_pretty(&value)?);
} else {
print_event(&value, fields.as_deref());
@ -436,6 +454,63 @@ async fn stream_events(
Ok(())
}
/// Tracks `id` -> `caused_by` for events seen so far in this stream/replay
/// batch, so each new event can be indented under its parent as it arrives
/// — no buffering, no waiting for the stream to end. An event whose parent
/// hasn't been seen yet (e.g. the parent predates a replay window, or the
/// causing event is filtered out by the subscription pattern) is rendered
/// as its own root rather than blocking on a parent that may never show up.
#[derive(Default)]
struct CausalityTracker {
parents: HashMap<String, Option<String>>,
}
impl CausalityTracker {
/// Depth = number of ancestors reachable by following `caused_by`.
/// Guards against cycles/self-loops (shouldn't happen, but a rendering
/// bug here must never hang the CLI) with both a seen-set and a hard cap.
fn depth(&self, id: &str) -> usize {
let mut depth = 0;
let mut current = id.to_string();
let mut seen = std::collections::HashSet::new();
while let Some(Some(parent)) = self.parents.get(&current) {
if depth >= 64 || !seen.insert(current.clone()) {
break;
}
depth += 1;
current = parent.clone();
}
depth
}
fn print(&mut self, event: &Value) {
let id = event.get("id").and_then(Value::as_str).map(str::to_string);
let caused_by = event
.get("caused_by")
.and_then(Value::as_str)
.map(str::to_string);
let depth = if let Some(id) = &id {
self.parents.insert(id.clone(), caused_by.clone());
self.depth(id)
} else {
0
};
let ts = event.get("timestamp").and_then(Value::as_u64).unwrap_or(0);
let event_name = event.get("event").and_then(Value::as_str).unwrap_or("?");
let source = event.get("source").and_then(Value::as_str).unwrap_or("?");
let time = format_timestamp(ts);
let indent = " ".repeat(depth);
let connector = if depth > 0 { "\u{2514}\u{2500} " } else { "" };
let id_display = id.as_deref().unwrap_or("?");
println!("{indent}{connector}{time} {event_name} source={source} id={id_display}");
if let Some(data) = event.get("data") {
println!("{indent} data: {data}");
}
}
}
fn print_json(value: &Value) -> Result<()> {
println!("{}", serde_json::to_string_pretty(value)?);
Ok(())

1
bread-shared/Cargo.toml
View file

@ -8,3 +8,4 @@ serde.workspace = true
serde_json.workspace = true
dirs.workspace = true
toml = "0.8"
uuid = { version = "1", features = ["v4"] }

75
bread-shared/src/lib.rs
View file

@ -90,20 +90,65 @@ pub struct BreadEvent {
pub source: AdapterSource,
/// Structured event data. The shape depends on the event family.
pub data: serde_json::Value,
/// Unique id for this specific event instance, assigned at construction.
///
/// *Since: v1.5* — enables causality tracking (`caused_by`) across chains
/// of Lua modules that re-emit events from inside `bread.on` handlers.
pub id: String,
/// The `id` of the event whose Lua handler emitted this event via
/// `bread.emit()`, if any.
///
/// `None` for events that originate outside any Lua handler invocation
/// (adapter-normalized events, IPC `emit`, daemon-internal sends like
/// `bread.system.startup` / `bread.profile.activated`). Populated only
/// when the event was constructed by `bread.emit()` while a subscriber
/// callback was synchronously running — see the "current dispatch id"
/// mechanism on `breadd`'s Lua engine.
///
/// *Since: v1.5*
pub caused_by: Option<String>,
}
impl BreadEvent {
/// Construct a new event with `timestamp` set to the current wall-clock.
/// Construct a new event with `timestamp` set to the current wall-clock,
/// a freshly generated `id`, and `caused_by` unset.
pub fn new(event: impl Into<String>, source: AdapterSource, data: serde_json::Value) -> Self {
Self::with_timestamp(event, now_unix_ms(), source, data)
}
/// Construct a new event with an explicit `timestamp`, preserving the
/// originating signal's observed time instead of "now". Used by the
/// normalizer, which carries `RawEvent::timestamp` through unchanged.
///
/// Like [`BreadEvent::new`], this always assigns a fresh `id` and leaves
/// `caused_by` unset — callers that need to thread causality set
/// `caused_by` on the returned value themselves.
pub fn with_timestamp(
event: impl Into<String>,
timestamp: u64,
source: AdapterSource,
data: serde_json::Value,
) -> Self {
Self {
event: event.into(),
timestamp: now_unix_ms(),
timestamp,
source,
data,
id: new_event_id(),
caused_by: None,
}
}
}
/// Generate a fresh unique id for a [`BreadEvent`].
///
/// Every construction path (`BreadEvent::new`, `BreadEvent::with_timestamp`,
/// and any remaining struct-literal construction) calls this so every event
/// gets a stable identity to hang `caused_by` chains off of.
pub fn new_event_id() -> String {
uuid::Uuid::new_v4().to_string()
}
/// Current Unix epoch in milliseconds.
///
/// Falls back to `0` if the system clock is before the epoch, which keeps
@ -316,6 +361,8 @@ mod tests {
timestamp: 1_700_000_000_000,
source: AdapterSource::Udev,
data: json!({ "id": "usb-1-1.4", "name": "Logitech" }),
id: "test-id-1".to_string(),
caused_by: Some("test-id-0".to_string()),
};
let raw = serde_json::to_string(&original).unwrap();
let decoded: BreadEvent = serde_json::from_str(&raw).unwrap();
@ -324,6 +371,30 @@ mod tests {
assert_eq!(decoded.timestamp, original.timestamp);
assert_eq!(decoded.source, original.source);
assert_eq!(decoded.data, original.data);
assert_eq!(decoded.id, original.id);
assert_eq!(decoded.caused_by, original.caused_by);
}
#[test]
fn bread_event_new_assigns_unique_id_and_no_cause() {
let a = BreadEvent::new("bread.test.a", AdapterSource::System, json!({}));
let b = BreadEvent::new("bread.test.b", AdapterSource::System, json!({}));
assert!(!a.id.is_empty());
assert_ne!(a.id, b.id, "each constructed event should get a unique id");
assert_eq!(a.caused_by, None);
}
#[test]
fn bread_event_with_timestamp_preserves_timestamp_and_assigns_id() {
let event = BreadEvent::with_timestamp(
"bread.test.c",
42,
AdapterSource::Udev,
json!({ "x": 1 }),
);
assert_eq!(event.timestamp, 42);
assert!(!event.id.is_empty());
assert_eq!(event.caused_by, None);
}
#[test]

56
breadd/src/core/normalizer.rs
View file

@ -43,6 +43,8 @@ impl EventNormalizer {
event: raw.kind.clone(),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}],
};
@ -143,6 +145,8 @@ impl EventNormalizer {
event: format!("bread.device.{}", verb),
timestamp: raw.timestamp,
source: AdapterSource::Udev,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"id": id,
"device": "unknown",
@ -173,36 +177,48 @@ impl EventNormalizer {
event: "bread.workspace.changed".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}],
"createworkspace" => vec![BreadEvent {
event: "bread.workspace.created".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "workspace": data }),
}],
"destroyworkspace" => vec![BreadEvent {
event: "bread.workspace.destroyed".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "workspace": data }),
}],
"monitoradded" => vec![BreadEvent {
event: "bread.monitor.connected".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "name": data }),
}],
"monitorremoved" => vec![BreadEvent {
event: "bread.monitor.disconnected".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "name": data }),
}],
"activewindow" => vec![BreadEvent {
event: "bread.window.focus.changed".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}],
"activewindowv2" => {
@ -211,6 +227,8 @@ impl EventNormalizer {
event: "bread.window.focused".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"address": fields.first().unwrap_or(&"")
}),
@ -222,6 +240,8 @@ impl EventNormalizer {
event: "bread.window.opened".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"address": fields.first().unwrap_or(&""),
"workspace": fields.get(1).unwrap_or(&""),
@ -236,6 +256,8 @@ impl EventNormalizer {
event: "bread.window.closed".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "address": fields.first().unwrap_or(&"") }),
}]
}
@ -245,6 +267,8 @@ impl EventNormalizer {
event: "bread.window.moved".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"address": fields.first().unwrap_or(&""),
"workspace": fields.get(1).unwrap_or(&""),
@ -255,6 +279,8 @@ impl EventNormalizer {
event: "bread.hyprland.event".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}],
}
@ -272,6 +298,8 @@ impl EventNormalizer {
},
timestamp: raw.timestamp,
source: AdapterSource::Power,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
});
}
@ -294,6 +322,8 @@ impl EventNormalizer {
event: event.to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Power,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
});
}
@ -304,6 +334,8 @@ impl EventNormalizer {
event: "bread.power.changed".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Power,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
});
}
@ -339,6 +371,8 @@ impl EventNormalizer {
event: "bread.device.connected".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"id": path,
"device": "unknown",
@ -352,6 +386,8 @@ impl EventNormalizer {
event: "bread.device.disconnected".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"id": path,
"device": "unknown",
@ -365,6 +401,8 @@ impl EventNormalizer {
event: "bread.bluetooth.device.paired".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"id": path,
"name": name,
@ -377,6 +415,8 @@ impl EventNormalizer {
event: "bread.bluetooth.device.unpaired".to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({
"id": path,
"address": address,
@ -421,6 +461,8 @@ impl EventNormalizer {
event: name.to_string(),
timestamp: raw.timestamp,
source: AdapterSource::Network,
id: bread_shared::new_event_id(),
caused_by: None,
data,
}]
}
@ -435,6 +477,8 @@ impl EventNormalizer {
event: format!("bread.terminal.{}", raw.kind),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}
@ -444,6 +488,8 @@ impl EventNormalizer {
event: format!("bread.remote.{}", raw.kind),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}
@ -453,6 +499,8 @@ impl EventNormalizer {
event: format!("bread.git.{}", raw.kind),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}
@ -462,6 +510,8 @@ impl EventNormalizer {
event: format!("bread.project.{}", raw.kind),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}
@ -474,6 +524,8 @@ impl EventNormalizer {
event: format!("bread.service.{suffix}"),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}
@ -490,6 +542,8 @@ impl EventNormalizer {
event,
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}
@ -512,6 +566,8 @@ impl EventNormalizer {
event: raw.kind.clone(),
timestamp: raw.timestamp,
source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(),
}]
}

2
breadd/src/core/state_engine.rs
View file

@ -656,6 +656,8 @@ mod tests {
timestamp: 0,
source: AdapterSource::System,
data,
id: bread_shared::new_event_id(),
caused_by: None,
}
}

2
breadd/src/ipc/mod.rs
View file

@ -27,7 +27,7 @@ use crate::lua::RuntimeHandle;
/// something new-but-additive (a binding, an event, an IPC param); bump the
/// major version only for a breaking change, which should not happen inside
/// this daemon's v1 lifetime per that section's stated policy.
const API_VERSION: &str = "1.4.0";
const API_VERSION: &str = "1.5.0";
#[derive(Clone)]
pub struct Server {

48
breadd/src/lua/mod.rs
View file

@ -211,6 +211,19 @@ struct LuaEngine {
next_sub_id: Arc<AtomicU64>,
next_timer_id: Arc<AtomicU64>,
current_module: Arc<Mutex<Option<String>>>,
/// The `id` of the `BreadEvent` whose subscriber callback is currently
/// executing, if any. Set immediately before invoking a handler in
/// `handle_event` and restored (not merely cleared) immediately after,
/// so nested/reentrant dispatch and a single event fan-out to multiple
/// subscriptions both see the correct parent id. Read synchronously by
/// `bread.emit()`'s binding to populate the outgoing event's
/// `caused_by`. The Lua engine runs single-threaded/cooperatively (one
/// engine instance processes `LuaMessage`s serially on a dedicated
/// thread), so a callback invocation and any `bread.emit()` inside it
/// happen synchronously within one `handle_event` call — no additional
/// synchronization beyond the existing `Mutex` (mirroring
/// `current_module` above) is needed.
current_dispatch_id: Arc<Mutex<Option<String>>>,
modules: Arc<Mutex<HashMap<String, ModuleInfo>>>,
module_decls: Arc<Mutex<HashMap<String, ModuleDecl>>>,
module_order: Arc<Mutex<Vec<String>>>,
@ -240,6 +253,7 @@ impl LuaEngine {
next_sub_id: Arc::new(AtomicU64::new(1)),
next_timer_id: Arc::new(AtomicU64::new(1)),
current_module: Arc::new(Mutex::new(None)),
current_dispatch_id: Arc::new(Mutex::new(None)),
modules: Arc::new(Mutex::new(HashMap::new())),
module_decls: Arc::new(Mutex::new(HashMap::new())),
module_order: Arc::new(Mutex::new(Vec::new())),
@ -438,6 +452,7 @@ impl LuaEngine {
bread.set("off", off_fn)?;
let emit_tx = self.emit_tx.clone();
let current_dispatch_id = self.current_dispatch_id.clone();
let emit_fn =
self.lua
.create_function(move |lua, (event_name, payload): (String, Value)| {
@ -447,8 +462,19 @@ impl LuaEngine {
.from_value::<serde_json::Value>(other)
.unwrap_or_else(|_| serde_json::json!({})),
};
// Daemon-internal emit — same trusted path as adapter/IPC
// construction, tagged System. If this runs synchronously
// inside a subscriber callback (see `handle_event`), thread
// the currently-dispatching event's id through as
// `caused_by` so the causality chain can be reconstructed.
let caused_by = current_dispatch_id
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone();
let mut event = BreadEvent::new(event_name, AdapterSource::System, data);
event.caused_by = caused_by;
emit_tx
.send(BreadEvent::new(event_name, AdapterSource::System, data))
.send(event)
.map_err(|_| LuaError::external("event channel closed"))?;
Ok(())
})?;
@ -1489,6 +1515,11 @@ impl LuaEngine {
}
self.set_current_module(module.clone());
// Every subscriber invocation triggered by this event should see the
// same parent id, and a `bread.emit()` call made synchronously
// inside the callback should attribute its `caused_by` to *this*
// event, not whatever was dispatching (if anything) before it.
let previous_dispatch_id = self.set_current_dispatch_id(Some(event.id.clone()));
let result = match kind {
HandlerKind::Event => {
let event_value = json_to_lua(&self.lua, &event)?;
@ -1502,6 +1533,9 @@ impl LuaEngine {
callback.call::<_, ()>((new_lua, old_lua))
}
};
// Restore rather than clear, so this correctly unwinds if dispatch
// is ever reentrant (e.g. a callback that pumps messages itself).
self.set_current_dispatch_id(previous_dispatch_id);
self.set_current_module(None);
if let Err(err) = result {
@ -1678,6 +1712,18 @@ impl LuaEngine {
}
}
/// Set the "currently dispatching" event id, returning whatever id was
/// there before. Callers restore the previous value (rather than
/// clearing to `None`) when the handler invocation finishes, so nested/
/// reentrant dispatch unwinds correctly — see `current_dispatch_id`'s
/// doc comment on the struct definition.
fn set_current_dispatch_id(&self, id: Option<String>) -> Option<String> {
match self.current_dispatch_id.lock() {
Ok(mut guard) => std::mem::replace(&mut *guard, id),
Err(poisoned) => std::mem::replace(&mut *poisoned.into_inner(), id),
}
}
fn cancel_all_timers(&self) {
if let Ok(mut map) = self.timers.lock() {
for (_, entry) in map.drain() {

142
breadd/tests/ipc_integration.rs
View file

@ -1,3 +1,4 @@
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
@ -521,6 +522,147 @@ async fn events_stream_receives_emitted_events() -> Result<()> {
Ok(())
}
/// A chain of 3 handlers, each reacting to the previous one's emitted event
/// entirely inside the Lua runtime (no test-side pumping between links):
/// module A handles the external trigger and emits X, module B subscribes
/// to X and emits Y, module C subscribes to Y and emits Z. This exercises
/// the "current dispatch id" mechanism in `breadd::lua::LuaEngine` end to
/// end: `caused_by` must thread through every hop of the chain, not just a
/// single emit.
#[tokio::test]
async fn event_causality_chain_threads_caused_by_across_handlers() -> Result<()> {
let harness = TestHarness::spawn_with_init(
r#"
-- module A: reacts to the external trigger, emits X
bread.on("bread.chain.trigger", function(event)
bread.emit("bread.chain.x", {})
end)
-- module B: reacts to X, emits Y
bread.on("bread.chain.x", function(event)
bread.emit("bread.chain.y", {})
end)
-- module C: reacts to Y, emits Z
bread.on("bread.chain.y", function(event)
bread.emit("bread.chain.z", {})
end)
"#,
)?;
harness.wait_until_ready().await?;
let stream = UnixStream::connect(harness.socket_path()).await?;
let (read_half, mut write_half) = stream.into_split();
let subscribe = json!({
"id": "sub-chain",
"method": "events.subscribe",
"params": { "filter": "bread.chain.*" }
});
write_half
.write_all(format!("{}\n", serde_json::to_string(&subscribe)?).as_bytes())
.await?;
let mut reader = BufReader::new(read_half).lines();
let ack = reader
.next_line()
.await?
.ok_or_else(|| anyhow!("missing subscribe ack"))?;
let ack_json: Value = serde_json::from_str(&ack)?;
assert_eq!(
ack_json
.get("result")
.and_then(|v| v.get("subscribed"))
.and_then(Value::as_bool),
Some(true)
);
// The trigger itself comes in over IPC's unsourced `emit`, outside any
// Lua handler — its `caused_by` must be None. Everything downstream
// (X, Y, Z) is emitted by `bread.emit()` from inside a running handler.
harness
.send_request("emit", json!({ "event": "bread.chain.trigger", "data": {} }))
.await?;
let mut events: HashMap<String, Value> = HashMap::new();
let deadline = Instant::now() + Duration::from_secs(5);
while events.len() < 4 && Instant::now() < deadline {
let Some(line) = reader.next_line().await? else {
break;
};
let event: Value = serde_json::from_str(&line)?;
let name = event
.get("event")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
if name.starts_with("bread.chain.") {
events.insert(name, event);
}
}
let trigger = events
.get("bread.chain.trigger")
.expect("missing trigger event");
let x = events.get("bread.chain.x").expect("missing X event");
let y = events.get("bread.chain.y").expect("missing Y event");
let z = events.get("bread.chain.z").expect("missing Z event");
let trigger_id = trigger
.get("id")
.and_then(Value::as_str)
.expect("trigger event missing id")
.to_string();
let x_id = x
.get("id")
.and_then(Value::as_str)
.expect("X event missing id")
.to_string();
let y_id = y
.get("id")
.and_then(Value::as_str)
.expect("Y event missing id")
.to_string();
let z_id = z
.get("id")
.and_then(Value::as_str)
.expect("Z event missing id")
.to_string();
assert_eq!(
trigger.get("caused_by").and_then(Value::as_str),
None,
"IPC-originated trigger event should have no caused_by"
);
assert_eq!(
x.get("caused_by").and_then(Value::as_str),
Some(trigger_id.as_str()),
"X should be caused_by the trigger event's id"
);
assert_eq!(
y.get("caused_by").and_then(Value::as_str),
Some(x_id.as_str()),
"Y should be caused_by X's id, not the original trigger"
);
assert_eq!(
z.get("caused_by").and_then(Value::as_str),
Some(y_id.as_str()),
"Z should be caused_by Y's id"
);
let ids: std::collections::HashSet<&str> =
[trigger_id.as_str(), x_id.as_str(), y_id.as_str(), z_id.as_str()]
.into_iter()
.collect();
assert_eq!(
ids.len(),
4,
"every event in the chain should have a distinct id"
);
harness.shutdown();
Ok(())
}
#[tokio::test]
async fn workflow_reaches_done_via_wait_any_happy_path() -> Result<()> {
let harness = TestHarness::spawn_with_init(