fix(pulse): harden GresIQ document dual-write rollout
Seto Elkahfi committed
Sep 4, 2026 at 17:51 UTC
f0d4016e8f0a344fb64a2786daea291aafaa6ec5
1 file changed
+143
-55
src/pulse/client.rs
+143
-55
@@ -40,21 +40,14 @@ impl PulseClient {
40
)
41
}
42
43
- /// ADR-0003 M2 dual-write toggle. When enabled (the default), every pulse
43
+ /// ADR-0003 M2 dual-write toggle. The mirror is deliberately opt-in while
44
+ /// document parity is being proven. When enabled, every attributed pulse
45
/// event is *also* written to the GresIQ document gateway
46
/// (`/gresiq/v1/collections/:collection`) in addition to the semantic
47
/// `pulse/*` routes, which still serve reads until the M5 cut-over. Set
47
- /// `ONDE_PULSE_DUAL_WRITE=0` (or `false`/`no`/`off`) to suppress it.
48
+ /// `ONDE_PULSE_DUAL_WRITE=1` (or `true`/`yes`/`on`) to enable it.
49
fn dual_write_enabled() -> bool {
49
- !matches!(
50
- std::env::var("ONDE_PULSE_DUAL_WRITE")
51
- .ok()
52
- .as_deref()
53
- .map(str::trim)
54
- .map(str::to_ascii_lowercase)
55
- .as_deref(),
56
- Some("0") | Some("false") | Some("no") | Some("off")
57
- )
50
+ dual_write_enabled_value(std::env::var("ONDE_PULSE_DUAL_WRITE").ok().as_deref())
51
}
52
53
/// Build a pulse client using the GresIQ credentials embedded in the SDK.
@@ -159,7 +152,7 @@ impl PulseClient {
152
if let Err(error) = client.inner.insert("pulse/model_loaded", &event).await {
153
log::warn!("pulse: model_loaded failed: {}", error);
154
}
162
- client.dual_write_model_loaded(&event).await;
155
+ client.spawn_model_loaded_dual_write(event);
156
};
157
158
if tokio::runtime::Handle::try_current().is_ok() {
@@ -249,68 +242,86 @@ impl PulseClient {
242
//
243
// These mirror each semantic pulse event into JSONB collections under the
244
// Onde gresiq_app. The relational `pulse/*` routes still own reads until the
252
- // M5 cut-over, so these writes are fire-and-forget: a failure is logged and
253
- // never affects inference. `onde_app_id` rides as a plain doc field — there
254
- // is no FK in the document store.
245
+ // M5 cut-over, so failures are logged and never affect inference.
246
+
247
+ /// Move the three model-load mirror requests off the model-loading critical
248
+ /// path. A dedicated thread is acceptable here because this runs once per
249
+ /// model activation, not once per inference, and also works for callers
250
+ /// whose original async runtime is short-lived.
251
+ fn spawn_model_loaded_dual_write(&self, event: ModelLoadedEvent) {
252
+ if !Self::dual_write_enabled() || event.onde_app_id.is_none() {
253
+ return;
254
+ }
255
+
256
+ let client = self.clone();
257
+ std::thread::spawn(move || {
258
+ let runtime = tokio::runtime::Builder::new_current_thread()
259
+ .enable_all()
260
+ .build();
261
+ match runtime {
262
+ Ok(runtime) => runtime.block_on(client.dual_write_model_loaded(&event)),
263
+ Err(error) => log::warn!(
264
+ "pulse: could not create runtime for model-loaded dual-write: {}",
265
+ error
266
+ ),
267
+ }
268
+ });
269
+ }
270
271
/// Mirror a model-load into `pulse_edges`, `pulse_models`, and
272
/// `pulse_deployments`. Each collection is a separate upsert (the
273
/// FK-ordered chain of the semantic route becomes client-orchestrated
274
/// multi-insert; the document store has no relationships to order).
275
async fn dual_write_model_loaded(&self, event: &ModelLoadedEvent) {
261
- if !Self::dual_write_enabled() {
276
+ let Some(onde_app_id) = event.onde_app_id.as_deref() else {
277
return;
263
- }
278
+ };
279
265
- self.write_doc(
266
- "pulse_edges",
267
- Some(&event.edge_id),
268
- &serde_json::json!({ "edge_id": event.edge_id }),
269
- )
270
- .await;
280
+ let edge_key = scoped_document_key(onde_app_id, &[&event.edge_id]);
281
272
- // One model assignment per app (the relational table is unique on
273
- // app_id); key by onde_app_id when present, else by the model itself.
274
- let model_key = event
275
- .onde_app_id
276
- .clone()
277
- .unwrap_or_else(|| event.model_id.clone());
278
- self.write_doc(
279
- "pulse_models",
280
- Some(&model_key),
281
- &serde_json::json!({
282
- "slug": event.model_id,
283
- "model_id": event.model_id,
284
- "model_name": event.model_name,
285
- "onde_app_id": event.onde_app_id,
286
- }),
287
- )
288
- .await;
282
+ let model_key = scoped_document_key(onde_app_id, &[&event.model_id]);
283
+ let deployment_key = scoped_document_key(onde_app_id, &[&event.edge_id, &event.model_id]);
284
+ let edge_document = edge_document(event, onde_app_id);
285
+ let model_document = serde_json::json!({
286
+ "slug": event.model_id,
287
+ "model_id": event.model_id,
288
+ "model_name": event.model_name,
289
+ "onde_app_id": event.onde_app_id,
290
+ });
291
+ let deployment_document = serde_json::json!({
292
+ "edge_id": event.edge_id,
293
+ "model_id": event.model_id,
294
+ "load_duration_ms": event.load_duration_ms,
295
+ "onde_app_id": event.onde_app_id,
296
+ "residency_state": "loaded",
297
+ });
298
290
- let deployment_key = format!("{}:{}", event.edge_id, event.model_id);
291
- self.write_doc(
292
- "pulse_deployments",
293
- Some(&deployment_key),
294
- &serde_json::json!({
295
- "edge_id": event.edge_id,
296
- "model_id": event.model_id,
297
- "load_duration_ms": event.load_duration_ms,
298
- "onde_app_id": event.onde_app_id,
299
- }),
300
- )
301
- .await;
299
+ tokio::join!(
300
+ self.write_doc("pulse_edges", Some(&edge_key), &edge_document),
301
+ self.write_doc("pulse_models", Some(&model_key), &model_document),
302
+ self.write_doc(
303
+ "pulse_deployments",
304
+ Some(&deployment_key),
305
+ &deployment_document,
306
+ ),
307
+ );
308
}
309
304
- /// Mirror an inference event into the append-only `pulse_inference_events`
305
- /// collection (no doc_key → server-generated UUID).
310
+ /// Mirror an inference event idempotently. Request IDs are already unique,
311
+ /// so using an app-scoped natural key prevents a retry from double-counting
312
+ /// the same inference during cutover validation.
313
async fn dual_write_inference(&self, event: &InferenceEvent) {
314
if !Self::dual_write_enabled() {
315
return;
316
}
317
+ let Some(onde_app_id) = event.onde_app_id.as_deref() else {
318
+ return;
319
+ };
320
+ let event_key = scoped_document_key(onde_app_id, &[&event.request_id]);
321
322
self.write_doc(
323
"pulse_inference_events",
313
- None,
324
+ Some(&event_key),
325
&serde_json::json!({
326
"edge_id": event.edge_id,
327
"model_id": event.model_id,
@@ -330,3 +341,80 @@ impl PulseClient {
341
}
342
}
343
}
344
+
345
+fn dual_write_enabled_value(value: Option<&str>) -> bool {
346
+ matches!(
347
+ value.map(str::trim).map(str::to_ascii_lowercase).as_deref(),
348
+ Some("1") | Some("true") | Some("yes") | Some("on")
349
+ )
350
+}
351
+
352
+/// Keys are scoped inside Onde's shared GresIQ app so identical installation
353
+/// ids belonging to different Onde customer apps cannot overwrite each other.
354
+fn scoped_document_key(onde_app_id: &str, parts: &[&str]) -> String {
355
+ std::iter::once(onde_app_id)
356
+ .chain(parts.iter().copied())
357
+ .collect::<Vec<_>>()
358
+ .join(":")
359
+}
360
+
361
+fn edge_document(event: &ModelLoadedEvent, onde_app_id: &str) -> serde_json::Value {
362
+ serde_json::json!({
363
+ "edge_id": event.edge_id,
364
+ "onde_app_id": onde_app_id,
365
+ "status": "online",
366
+ })
367
+}
368
+
369
+#[cfg(test)]
370
+mod tests {
371
+ use super::*;
372
+
373
+ #[test]
374
+ fn dual_write_requires_an_explicit_positive_flag() {
375
+ for disabled in [
376
+ None,
377
+ Some(""),
378
+ Some("0"),
379
+ Some("false"),
380
+ Some("no"),
381
+ Some("off"),
382
+ ] {
383
+ assert!(!dual_write_enabled_value(disabled));
384
+ }
385
+ for enabled in [Some("1"), Some("true"), Some("YES"), Some(" on ")] {
386
+ assert!(dual_write_enabled_value(enabled));
387
+ }
388
+ }
389
+
390
+ #[test]
391
+ fn document_keys_include_the_onde_app_boundary() {
392
+ assert_eq!(
393
+ scoped_document_key("app-a", &["edge-1", "model-1"]),
394
+ "app-a:edge-1:model-1"
395
+ );
396
+ assert_ne!(
397
+ scoped_document_key("app-a", &["onde-unknown"]),
398
+ scoped_document_key("app-b", &["onde-unknown"])
399
+ );
400
+ assert_eq!(
401
+ scoped_document_key("app-a", &["onde-1720000000000-42"]),
402
+ "app-a:onde-1720000000000-42"
403
+ );
404
+ }
405
+
406
+ #[test]
407
+ fn edge_document_carries_cutover_state_and_tenant_id() {
408
+ let event = ModelLoadedEvent {
409
+ edge_id: "edge-1".to_string(),
410
+ model_id: "model-1".to_string(),
411
+ model_name: "Model 1".to_string(),
412
+ load_duration_ms: 42,
413
+ onde_app_id: Some("app-a".to_string()),
414
+ };
415
+ let document = edge_document(&event, "app-a");
416
+ assert_eq!(document["edge_id"], "edge-1");
417
+ assert_eq!(document["onde_app_id"], "app-a");
418
+ assert_eq!(document["status"], "online");
419
+ }
420
+}