Stop dropped streams and key inference docs by edge
When the stream receiver is dropped, the send fails and the loop used to keep generating, then reported the inference as "success". It now stops and reports "cancelled". The inference document key was app_id:request_id. Request IDs are a millisecond plus a per-process counter, so two edges of the same app can produce the same ID and overwrite each other's document. The key now includes the edge_id. Co-Authored-By: siGit Code <noreply@sigit.si>
Seto Elkahfi committed
Sep 23, 2026 at 22:39 UTC
7bde72e9c3d6750c265725833bb4113e4fb678df
3 files changed
+21
-9
src/inference/engine.rs
+8
-1
@@ -1258,13 +1258,20 @@ impl ChatEngine {
1258
started.elapsed().as_millis() as u64
1259
});
1260
assembled.push_str(text);
1261
- let _ = tx
1261
+ let sent = tx
1262
.send(StreamChunk {
1263
delta: text.clone(),
1264
done: false,
1265
finish_reason: None,
1266
})
1267
.await;
1268
+ // The receiver is gone, so nobody is
1269
+ // reading the rest. Stop generating and
1270
+ // keep only what the caller received.
1271
+ if sent.is_err() {
1272
+ status = "cancelled";
1273
+ break;
1274
+ }
1275
}
1276
if let Some(ref reason) = choice.finish_reason {
1277
last_finish_reason = Some(reason.clone());
src/pulse/client.rs
+11
-7
@@ -309,9 +309,11 @@ impl PulseClient {
309
);
310
}
311
312
- /// Mirror an inference event idempotently. Request IDs are already unique,
313
- /// so using an app-scoped natural key prevents a retry from double-counting
314
- /// the same inference during cutover validation.
312
+ /// Mirror an inference event idempotently, so a retry can't double-count
313
+ /// the same inference during cutover validation. Request IDs are only
314
+ /// unique within one process (millisecond plus a per-process counter), so
315
+ /// two edges of the same app can mint the same ID; the edge is part of the
316
+ /// key for that reason.
317
async fn dual_write_inference(&self, event: &InferenceEvent) {
318
if !Self::dual_write_enabled() {
319
return;
@@ -319,7 +321,7 @@ impl PulseClient {
321
let Some(onde_app_id) = event.onde_app_id.as_deref() else {
322
return;
323
};
322
- let event_key = scoped_document_key(onde_app_id, &[&event.request_id]);
324
+ let event_key = scoped_document_key(onde_app_id, &[&event.edge_id, &event.request_id]);
325
let document = inference_document(event);
326
327
self.write_doc("pulse_inference_events", Some(&event_key), &document)
@@ -426,9 +428,11 @@ mod tests {
428
scoped_document_key("app-a", &["onde-unknown"]),
429
scoped_document_key("app-b", &["onde-unknown"])
430
);
429
- assert_eq!(
430
- scoped_document_key("app-a", &["onde-1720000000000-42"]),
431
- "app-a:onde-1720000000000-42"
431
+ // Request IDs collide across edges (each process counts from 0), so
432
+ // the inference key carries the edge.
433
+ assert_ne!(
434
+ scoped_document_key("app-a", &["edge-1", "onde-1720000000000-0"]),
435
+ scoped_document_key("app-a", &["edge-2", "onde-1720000000000-0"])
436
);
437
}
438
src/pulse/events.rs
+2
-1
@@ -40,7 +40,8 @@ pub struct InferenceEvent {
40
/// `"success"`, `"cancelled"`, or `"error"`. A blocking request only
41
/// reports `"success"`, since its errors return before reaching pulse. A
42
/// stream can fail after it starts, so a mid-stream model error reports
43
- /// `"error"`, usually with no `ttft_ms`.
43
+ /// `"error"`, usually with no `ttft_ms`, and a stream whose receiver was
44
+ /// dropped reports `"cancelled"`.
45
pub status: String,
46
/// Same Onde app as the corresponding `ModelLoadedEvent`. `None` for SDK
47
/// builds without app credentials or direct Rust consumers.