| 322 | } |
| 323 | |
| 324 | void DataSourceRuntimeHost::setObjectStoreTarget(ObjectStore* target) { |
| 325 | // Route cbPushMessage's lazy-object push through the swap (it pushed straight |
| 326 | // to the primary before, evicting the paused scrub-back snapshot). |
| 327 | object_store_target_.store(target); |
| 328 | // TODO(stream-pause): deferred edge cases (none hit a topic registered before |
| 329 | // the first pause): bindings created while paused still init against A; an |
| 330 | // in-flight push can strand a frame in B across the resume flush; resume's |
| 331 | // catch-up notifyIngest lists only scalar TopicIds (object-only flush nudge). |
| 332 | // Retarget the source-level object write host and every per-parser-binding |
| 333 | // one (the streaming hot path). Each host's atomic swap lets in-flight |
| 334 | // pushes finish on the old target while the next lands on the new one; the |
| 335 | // manager guarantees `target` already has the topics (lockstep mirror). |
| 336 | source_object_write_host_.setTarget(target); |
| 337 | for (auto& [_id, binding] : parser_bindings_) { |
| 338 | if (binding.object_write_host != nullptr) { |
| 339 | binding.object_write_host->setTarget(target); |
| 340 | } |
| 341 | } |
| 342 | } |
| 343 | |
| 344 | void DataSourceRuntimeHost::setDataEngineTarget(DataEngine* target) { |
| 345 | // Mirrors setObjectStoreTarget but for scalar writes. Retargets the |