llo/dev/v31: move blob broadcasting out of Observation - #256
Merged
Conversation
Stream values are now gathered by a background blob pump (blobpump.go) that holds the BlobBroadcastFetcher provided to the factory. Each cycle observes the streams and seqNr known at the time, serializes the values, broadcasts them, and parks the marshaled handle. Observation keeps only the cheap synchronous work (KV load, retirement, channel votes), publishes the round context, and picks up whatever the pump has ready. Cadence is consumption driven: a cycle is kicked whenever Observation takes or discards a snapshot, so the pump rate tracks the round rate without the plugin needing to know OCR round cadence, and no blob is broadcasted unnecessarily. Cycles are serial so only one Observe is ever in flight. There is no idle watchdog: snapshot usability is bounded by sequence number, so refreshing while no rounds run would only produce snapshots already too old to use. Stream values are now carried exclusively by blobs and never inline. A round that finds nothing usable (cold start, failed cycle, stale snapshot) emits an observation carrying only votes and its timestamp. Adds llo/dev/v31/llotest, exporting an in-memory, content-addressed BlobBroadcastFetcher that makes the broadcast/fetch/merge round trip testable.
product-security-plaid-production
Bot
requested a review
from sebastian-lerner-link
August 19, 2026 18:24
|
👋 brunotm, thanks for creating this pull request! To help reviewers, please consider creating future PRs as drafts first. This allows you to self-review and make any final changes before notifying the team. Once you're ready, you can mark it as "Ready for review" to request feedback. Thanks! |
There was a problem hiding this comment.
Pull request overview
Moves stream observation and blob broadcasting off the OCR critical path into a background pump.
Changes:
- Adds consumption-driven asynchronous blob pumping.
- Makes stream values blob-only and updates factory configuration.
- Adds an in-memory blob test double and pump coverage.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
llo/dev/v31/plugin.go |
Consumes prepared blob snapshots. |
llo/dev/v31/plugin_test.go |
Updates observation tests. |
llo/dev/v31/observation.go |
Removes inline value encoding. |
llo/dev/v31/llotest/blob.go |
Adds in-memory blob transport. |
llo/dev/v31/llotest/blob_test.go |
Tests blob transport. |
llo/dev/v31/flow_test.go |
Tests asynchronous plugin flow. |
llo/dev/v31/factory.go |
Configures and starts the pump. |
llo/dev/v31/doc.go |
Documents blob-pump behavior. |
llo/dev/v31/blobpump.go |
Implements the background pump. |
llo/dev/v31/blobpump_test.go |
Tests pump lifecycle and failures. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
MaxPerOracleUnexpiredBlobCount and its cumulative payload counterpart were hardcoded to 32, independent of the configurable BlobLifetimeRounds. Since the pump broadcasts about one blob per round, a lifetime above 32 could leave more than 32 blobs unexpired, at which point libocr rejects further broadcasts for resource exhaustion and rounds continually lose stream values. Both limits are now derived from the configured lifetime plus BlobReapingMarginRounds (asynchronous reaping), floored at the previous value of 32. BlobLifetimeRounds above MaxBlobLifetimeRounds (64) is rejected at plugin construction so the declared budget stays bounded. The default lifetime of 3 keeps the previous limits unchanged.
justinfranco
approved these changes
Aug 19, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stream values are now gathered by a background blob pump with the BlobBroadcastFetcher provided to the factory. Each cycle observes the streams and seqNr known at the time, serializes the values, broadcasts them, and parks the marshaled handle for the next observation. Observation keeps only the cheap synchronous work (KV load, retirement, channel votes), publishes the round context, and picks up the handles prepared by the pump.
Cadence is consumption driven, a cycle is kicked whenever Observation takes or discards a snapshot, so the pump rate tracks the round rate without the plugin needing to know OCR round cadence, and no blob is broadcasted unnecessarily. Pump cycles are serial so only one is ever in flight. There is no idle watchdog, snapshot usability is bounded by sequence number, so refreshing while no rounds run would only produce snapshots already too old to use.
Stream values are now carried exclusively by blobs and never inline.
A round that finds nothing usable (cold start, failed cycle, stale snapshot) emits an observation carrying only votes and its timestamp.
Adds llo/dev/v31/llotest, exporting an in-memory, content-addressed BlobBroadcastFetcher that makes the broadcast/fetch/merge round trip testable.