| 308 | } |
| 309 | |
| 310 | void DataSourceRuntimeHost::setObjectRetentionBudget(int64_t time_window_ns, size_t max_memory_bytes) { |
| 311 | // Budget the active store (B while paused) so the paused tail stays bounded; |
| 312 | // the frozen store is untouched (eviction is push-triggered anyway). |
| 313 | ObjectStore* target = object_store_target_.load(); |
| 314 | for (auto& [_id, binding] : parser_bindings_) { |
| 315 | if (!binding.object_topic_id.has_value()) { |
| 316 | continue; |
| 317 | } |
| 318 | target->setRetentionBudget( |
| 319 | *binding.object_topic_id, |
| 320 | RetentionBudget{.time_window_ns = time_window_ns, .max_memory_bytes = max_memory_bytes}); |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | void DataSourceRuntimeHost::setObjectStoreTarget(ObjectStore* target) { |
| 325 | // Route cbPushMessage's lazy-object push through the swap (it pushed straight |