Isolate cached feed broadcast consumers #92

goodboy · 2026-08-14T00:29:12Z

goodboy commented

2026-08-14 00:29:12 +00:00

Isolate cached feed broadcast consumers

Motivation

Several quote consumers treated a cached stream as both the shared source and a caller-owned receive cursor. Cache hits could construct competing broadcaster roots, overwrite shared Flume.stream fields, or mutate private lag policy on a receiver owned by another task.

That split quotes between callers and let one caller’s teardown or lag policy leak into its peers. This branch keeps one root per cached resource and gives every consumer a lexical child subscription.

Src of research

The public subscription and lifecycle contracts are provided by the paired tractor broadcast hardening work,

tractor PR #490

Summary of changes

Give clear_dark_triggers() a dedicated non-strict quote subscription instead of mutating MsgStream._raise_on_lag.
Keep provider roots private to cached Feed resources and return caller-local Feed and Flume views keyed by the complete symbol set and stream policy.
- Subscribe every multi-provider relay before yielding its merged channel.
- Give each relay an independent send-channel clone so one provider EOC cannot terminate sibling relays.
Make Deribit price-feed misses and hits own children from one cached LinkedTaskChannel broadcaster root.
Replace IB’s manual _quote_streams registry with a symbol-keyed cached channel context and caller-owned child subscriptions.
Add deterministic ownership regressions for EMS, feed-cache, Deribit and IB fan-out, teardown and cache identity.

TODOs before landing

Land tractor PR #490 so piker can use the public subscriber lag-policy and root-lifecycle contracts exercised here.

Future follow up

Add a public duplex child-stream context interface in tractor. piker.data.feed.maybe_open_feed() still grafts MsgStream.ctx onto BroadcastReceiver._ctx for pause/resume control compatibility.

Links

tractor PR #490 - required broadcast API and lifecycle hardening.

(this pr content was generated in some part by opencode using gpt-5.6-sol (openai))

## Isolate cached feed broadcast consumers ### Motivation Several quote consumers treated a cached stream as both the shared source and a caller-owned receive cursor. Cache hits could construct competing broadcaster roots, overwrite shared `Flume.stream` fields, or mutate private lag policy on a receiver owned by another task. That split quotes between callers and let one caller's teardown or lag policy leak into its peers. This branch keeps one root per cached resource and gives every consumer a lexical child subscription. --- ### Src of research The public subscription and lifecycle contracts are provided by the paired `tractor` broadcast hardening work, - [tractor PR #490][tractor-490] --- ### Summary of changes - Give `clear_dark_triggers()` a dedicated non-strict quote subscription instead of mutating `MsgStream._raise_on_lag`. - Keep provider roots private to cached `Feed` resources and return caller-local `Feed` and `Flume` views keyed by the complete symbol set and stream policy. * Subscribe every multi-provider relay before yielding its merged channel. * Give each relay an independent send-channel clone so one provider EOC cannot terminate sibling relays. - Make Deribit price-feed misses and hits own children from one cached `LinkedTaskChannel` broadcaster root. - Replace IB's manual `_quote_streams` registry with a symbol-keyed cached channel context and caller-owned child subscriptions. - Add deterministic ownership regressions for EMS, feed-cache, Deribit and IB fan-out, teardown and cache identity. --- ### TODOs before landing - [ ] Land [tractor PR #490][tractor-490] so piker can use the public subscriber lag-policy and root-lifecycle contracts exercised here. --- ### Future follow up - [ ] Add a public duplex child-stream context interface in `tractor`. `piker.data.feed.maybe_open_feed()` still grafts `MsgStream.ctx` onto `BroadcastReceiver._ctx` for pause/resume control compatibility.   ### Links - [tractor PR #490][tractor-490] - required broadcast API and lifecycle hardening. --- (this pr content was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) [tractor-490]: https://github.com/goodboy/tractor/pull/490

goodboy added 4 commits 2026-08-14 00:29:12 +00:00

12b0d66e85 .clearing: own dark quote broadcast subscription

Give `clear_dark_triggers()` a dedicated
`MsgStream.subscribe()` handle with `raise_on_lag=False` instead
of mutating the private lag policy on whichever root or cached
child sits in `Flume.stream`.

Also,
- factor the hot quote loop into `_clear_dark_triggers()` so the
  public task owns the child receiver for its full lifetime.
- add an EMS regression proving the lag policy, child lifetime,
  and source-stream state remain isolated.

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

b691422b26 .data: isolate cached feed stream consumers

Keep provider root streams private to the cached `Feed` and give every
`maybe_open_feed()` caller, including the first owner, a local
subscription view with copied `Flume` descriptors.

Also,
- key cached resources by the normalized full symbol set and stream
  policy instead of only the first FQME.
- subscribe each `Feed.open_multi_stream()` provider relay and wait for
  all relays to start before yielding the merged channel.
- give each provider relay an independent send-channel clone so one
  provider EOC cannot terminate its siblings.
- document `Feed.pause()` and `Feed.resume()` as shared remote-provider
  controls rather than caller-local cursor operations.
- cover miss/hit ownership, multi-market provider mapping, relay
  fan-out, lexical teardown and cache-key separation.

Prompt-IO: ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

ddac3553a3 .deribit: own cached price-feed subscriptions

Give every `maybe_open_price_feed()` caller a lexical child from the
cached `LinkedTaskChannel` broadcaster instead of yielding the raw
channel to the first owner and constructing competing roots for hits.

Add a concurrent regression proving miss and hit callers receive the
same quote, close independently, and leave the retained root alive.

Prompt-IO: ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

4af885d90f .ib.feed: cache quote channel contexts

Replace the manual `_quote_streams` raw-channel registry with a
symbol-keyed `maybe_open_context()` resource whose lifetime extends to
the final quote consumer.

Give every `open_aio_quote_stream()` caller a lexical child from the
cached `LinkedTaskChannel` broadcaster so concurrent users receive the
same quotes and close independently.

Prompt-IO: ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

goodboy changed title from ~~wkt/fix_broadcast_consumers~~ to Isolate cached feed broadcast consumers

2026-08-14 00:37:13 +00:00

This pull request can be merged automatically.

You are not authorized to merge this pull request.

You can also view command line instructions.

Step 1:

From your project repository, check out a new branch and test the changes.

git checkout -b wkt/fix_broadcast_consumers datad_service

git pull origin wkt/fix_broadcast_consumers

Step 2:

Merge the changes and update on Gitea.

git checkout datad_service

git merge --no-ff wkt/fix_broadcast_consumers

git push origin datad_service

Sign in to join this conversation.

No reviewers

No Label

No Milestone

No project

No Assignees

1 Participants

Notifications

Due Date

The due date is invalid or out of range. Please use the format 'yyyy-mm-dd'.

No due date set.

Dependencies

No dependencies set.

Reference: pikers/piker#92