| 425 | } |
| 426 | |
| 427 | void StreamingSourceManager::workerLoop(DatasetId dataset_id) { |
| 428 | auto it = sessions_.find(dataset_id); |
| 429 | if (it == sessions_.end()) { |
| 430 | return; |
| 431 | } |
| 432 | StreamingSession* sess = it->second.get(); |
| 433 | |
| 434 | while (!sess->runtime_host->stopRequested()) { |
| 435 | if (auto status = sess->handle.poll(); !status) { |
| 436 | const QString err = QString::fromStdString(status.error()); |
| 437 | QMetaObject::invokeMethod( |
| 438 | this, [this, dataset_id, err]() { emit streamError(dataset_id, err); }, Qt::QueuedConnection); |
| 439 | sess->runtime_host->requestStop(std::string("plugin poll error: ") + err.toStdString()); |
| 440 | break; |
| 441 | } |
| 442 | sess->runtime_host->flushPending(); |
| 443 | // Broadcast live-advance on the UI thread so plot adapters invalidate |
| 444 | // their sample caches AND follow-live consumers auto-fit. The write |
| 445 | // host wrote directly through DataEngine and bypassed SessionManager's |
| 446 | // own commitChunks → neither Qt signal would fire otherwise. |
| 447 | // |
| 448 | // The `live` flag is gated on `paused_`: when the user has paused the |
| 449 | // stream, samples keep being committed but the plot stops following the |
| 450 | // live edge (PlotWidget skips resetZoom and just replots in place). |
| 451 | // |
| 452 | // Retention enforcement piggybacks on the same UI-thread hop so |
| 453 | // `retention_seconds_` stays UI-thread-only (no atomic needed). Trim |
| 454 | // runs before notifyIngest so consumers observe the post-trim engine. |
| 455 | QMetaObject::invokeMethod( |
| 456 | this, |
| 457 | [this, dataset_id]() { |
| 458 | const auto window_ns = static_cast<int64_t>(retention_seconds_) * 1'000'000'000LL; |
| 459 | // Apply the budget on every tick: cheap (one map iteration per |
| 460 | // session) and idempotent. Solves the cold-start problem where the |
| 461 | // plugin creates new object topics mid-stream (post-startSession) |
| 462 | // and would otherwise miss the budget set by setObjectRetentionBudget. |
| 463 | if (auto session_it = sessions_.find(dataset_id); session_it != sessions_.end()) { |
| 464 | session_it->second->runtime_host->setObjectRetentionBudget(window_ns, kStreamingObjectMemoryBudget); |
| 465 | } |
| 466 | // Run eager filters over the freshly committed input BEFORE retention |
| 467 | // can evict it (the streaming correctness order, plan §5/D6): a stateful |
| 468 | // node must consume every sample before it ages out of the window. On |
| 469 | // the UI thread here, serialized with the Filter Editor's own |
| 470 | // apply/update and with the retention trim just below. While paused the |
| 471 | // writes land in the secondary engine (no DerivedEngine); the resume |
| 472 | // catch-up advances the whole buffered tail before re-trimming. |
| 473 | if (!paused_) { |
| 474 | const auto filter_inputs = session_manager_.createReader().listTopics(dataset_id); |
| 475 | (void)session_manager_.dataProcessorService().advanceOnCommit(filter_inputs); |
| 476 | } |
| 477 | |
| 478 | // Trim the engine currently being written: B while paused (bounds the |
| 479 | // tail), the primary while live. The frozen engine is never touched, |
| 480 | // preserving the primary snapshot for scrub-back. Scope the trim to THIS |
| 481 | // streaming dataset so a file the user loaded into the same engine keeps |
| 482 | // its full history. |
| 483 | if (paused_) { |
| 484 | secondary_data_engine_->enforceRetention(window_ns, dataset_id); |
nothing calls this directly
no test coverage detected