Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **A market-by-price `InstrumentReset` discarded its own recovery snapshot group.** A publisher withholding and then re-admitting a market emits two resets, and the re-admission's recovery `SnapshotBegin` follows its own reset by ~0.1 ms on a different port — resets ride mktdata, snapshot groups the snapshot port, on independent sequence series — so which is processed first is a coin flip. When the begin won, both the book's assembly and the processor's level route were torn out, the market waited a full snapshot rotation to heal, and the group's levels were counted on `dz_mbp_snapshot_levels_dropped_total{reason="reset"}`. A group anchored at or after the reset's `New Anchor Seq` is now kept — the same test a `SnapshotBegin` is already judged by — and only one that predates the reset is dropped and tombstoned. Observed on a live Phoenix feed at roughly two flaps per hour per market; expect that `reason="reset"` series to fall substantially.

### Changed
- **`edge-kalshi-sports-mbp`'s `category` is now `events`, and its group `code` is unchanged.** The
universe that row carries is event markets, of which sport is one kind, so `sports` named it too
narrowly; the group name is the ledger's and a subscriber matches `doublezero status` on it, so
renaming that here would have darkened the row silently. Nothing an operator passes changes:
`--channels`/`DZ_CHANNELS` key on the `code`, not the category. What does change is the
producer-side identity every universe-scoped key is built from — the arbitration scope
(`(venue, category)`), the tape-owner rank, the catalog, book and history keys — plus the two
read-only surfaces that report it: `category` in `/v1/status`'s `channels.rows[]` and in
`/admin/channels`, and the `(category: …)` suffix `/v1/products` appends when a symbol resolves in
more than one universe under one Source ID. `category` is producer-side only and is not
serialized onto the WebSocket, so PROTOCOL.md is untouched.

⚠️ **The hosted feed-registry document has to be republished for any of that to take effect.**
The image bakes `DZ_FEED_REGISTRY_URL` in and a URL origin wins over the compiled-in copy, so on a
default container `src/ingest/registry.json` is the fallback, not the authority — until the
document served at that URL carries `"category": "events"`, every surface above keeps reporting
`sports` and the rename is invisible to an operator. Same shape as the stale `lashay-*` group
codes below, and the document is behind in the same way today: it carries neither the `elections`
rows nor the Phoenix row. Clearing `DZ_FEED_REGISTRY_URL` on one host makes the built-in document
win, which is the workaround until it is republished.
- **The feed registry's `notes` are written for whoever runs the bridge now, not for whoever wrote
the row.** Every row carried a different mixture of what a subscriber needs and how we came to
know it — dated captures, port allocations that were once wrong and have since been corrected,
Expand Down
4 changes: 3 additions & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,9 @@ Modules are grouped by role under `src/`:
no inventory, and its depth half is the group's highest-volume publisher — its owner is still to
be established, so **do not drop the row** for being unattributed. A third **Kalshi** row
(`edge-kalshi-sports-mbp`, group `233.84.178.20`, MBP, `Sticky`, claiming the tape) carries a *disjoint* universe
under the same Source ID — hence its own `category` — and is the one `derived` row: 31 channels
under the same Source ID — hence its own `category` (`events`: sport is one kind of event
market and the row carries more than sport, while the group `code` stays the ledger's
`edge-kalshi-sports-mbp`) — and is the one `derived` row: 31 channels
(ids 10-29, 39-48, 49) expanded to `34000`/`44000`/`54000 + id`, confirmed against the publishers'
own deployment inventory on 2026-08-09. `33000`/`43000` is the **top-of-book sibling's** base, and a
market-by-price row built on it joins the right group and receives nothing, which reads as a dead
Expand Down
4 changes: 2 additions & 2 deletions src/ingest/channel_filter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -393,7 +393,7 @@ mod tests {
/// excluded channel either way, so that assertion could not fail.
#[test]
fn an_excluded_sports_channel_binds_no_socket() {
let feed = feeds().iter().find(|f| f.category == "sports").unwrap();
let feed = feeds().iter().find(|f| f.category == "events").unwrap();
let f = ChannelFilter::parse("edge-kalshi-sports-mbp=10,11").unwrap();
let ports: Vec<u16> = f
.publishers_for(feed)
Expand All @@ -407,7 +407,7 @@ mod tests {
/// filtering a feed nobody narrowed: an unmentioned feed binds its whole published set.
#[test]
fn an_unmentioned_feed_binds_every_publisher() {
let feed = feeds().iter().find(|f| f.category == "sports").unwrap();
let feed = feeds().iter().find(|f| f.category == "events").unwrap();
let f = ChannelFilter::parse("").unwrap();
assert_eq!(f.publishers_for(feed).len(), feed.publishers.len());
assert!(
Expand Down
6 changes: 3 additions & 3 deletions src/ingest/feeds.rs
Original file line number Diff line number Diff line change
Expand Up @@ -368,8 +368,8 @@ mod tests {
fn sports_ports_are_the_base_plus_the_channel_id() {
let row = feeds()
.iter()
.find(|f| f.venue == "KALSHI" && f.category == "sports")
.expect("no sports row");
.find(|f| f.venue == "KALSHI" && f.category == "events")
.expect("no events row");
assert_eq!(row.code, "edge-kalshi-sports-mbp");
assert_eq!(row.group, Ipv4Addr::new(233, 84, 178, 20));
assert_eq!(row.kind, FeedKind::MarketByPrice);
Expand Down Expand Up @@ -459,7 +459,7 @@ mod tests {
("PHOENIX", "perps", FeedKind::MarketByPrice) => "edge-phoenix-mbp",
("KALSHI", "perps", FeedKind::TopOfBook) => "edge-kalshi-perps-tob",
("KALSHI", "perps", FeedKind::MarketByPrice) => "edge-kalshi-perps-mbp",
("KALSHI", "sports", FeedKind::MarketByPrice) => "edge-kalshi-sports-mbp",
("KALSHI", "events", FeedKind::MarketByPrice) => "edge-kalshi-sports-mbp",
("KALSHI", "elections", FeedKind::TopOfBook) => "edge-kalshi-elections-pol-tob",
("KALSHI", "elections", FeedKind::MarketByPrice) => "edge-kalshi-elections-pol-mbp",
other => panic!("unexpected feed {other:?}"),
Expand Down
56 changes: 28 additions & 28 deletions src/ingest/reconcile.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1407,7 +1407,7 @@ mod tests {
#[test]
fn disjoint_categories_each_hold_their_own_tape() {
let perps_tob = ("KALSHI", "perps", TOB, 7576);
let sports_mbp = ("KALSHI", "sports", MBP, 34010);
let sports_mbp = ("KALSHI", "events", MBP, 34010);
let owners = tape_owners([perps_tob, sports_mbp], all_live);
assert!(owns(&owners, &perps_tob));
assert!(
Expand Down Expand Up @@ -1521,8 +1521,8 @@ mod tests {
fn the_filter_narrows_the_desired_receiver_set() {
let sports = *crate::ingest::feeds::feeds()
.iter()
.find(|f| f.category == "sports")
.expect("the built-in registry has a sports row");
.find(|f| f.category == "events")
.expect("the built-in registry has an events row");
let filter = ChannelFilter::parse("edge-kalshi-sports-mbp=10,11").unwrap();

let narrowed = test_reconciler_with_filter(vec![sports], filter);
Expand All @@ -1543,14 +1543,14 @@ mod tests {
);
}

/// The real "sports" row (group code `edge-kalshi-sports-mbp`), for tests that need genuine channel-filter
/// The real "events" row (group code `edge-kalshi-sports-mbp`), for tests that need genuine channel-filter
/// narrowing — `ChannelFilter::parse` validates against the loaded registry, so a custom `Feed`
/// with a made-up code cannot be narrowed at all.
fn sports_row() -> Feed {
*crate::ingest::feeds::feeds()
.iter()
.find(|f| f.category == "sports")
.expect("the built-in registry has a sports row")
.find(|f| f.category == "events")
.expect("the built-in registry has an events row")
}

/// One `book` batch for a given identity, reused by the tests below that seed a real arbiter.
Expand Down Expand Up @@ -1658,11 +1658,11 @@ mod tests {
vec![sports_row()],
ChannelFilter::parse("edge-kalshi-sports-mbp=10,11").unwrap(),
);
let key10 = ("KALSHI", "sports", FeedKind::MarketByPrice, 34010u16);
let key10 = ("KALSHI", "events", FeedKind::MarketByPrice, 34010u16);

let hist_key = history::Key {
source_id: 3,
category: "sports".into(),
category: "events".into(),
channel: 10,
instrument_id: 1,
};
Expand Down Expand Up @@ -1923,7 +1923,7 @@ mod tests {
);

r.cfg.instruments.lock().unwrap().insert(
("KALSHI".into(), "sports".into(), 10u8, 1u32),
("KALSHI".into(), "events".into(), 10u8, 1u32),
crate::model::NormalizedInstrument {
tick_size: 0,
venue: "KALSHI".into(),
Expand All @@ -1932,15 +1932,15 @@ mod tests {
symbol: "DEPARTED".into(),
channel: 10,
instrument_id: 1,
category: "sports".into(),
category: "events".into(),
price_exponent: -4,
qty_exponent: -2,
},
);
seed_book(&r, "KALSHI", 3, "DEPARTED", "sports", 10, 1);
seed_book(&r, "KALSHI", 3, "DEPARTED", "events", 10, 1);
let hist_key = history::Key {
source_id: 3,
category: "sports".into(),
category: "events".into(),
channel: 10,
instrument_id: 1,
};
Expand Down Expand Up @@ -2040,14 +2040,14 @@ mod tests {
vec![feed],
ChannelFilter::parse("edge-kalshi-sports-mbp=10,11").unwrap(),
);
let key10 = ("KALSHI", "sports", FeedKind::MarketByPrice, 34010u16);
let key10 = ("KALSHI", "events", FeedKind::MarketByPrice, 34010u16);
let catalog_key: (Arc<str>, Arc<str>, u8, u32) =
("KALSHI".into(), "sports".into(), 10u8, 1u32);
("KALSHI".into(), "events".into(), 10u8, 1u32);
let book_key: crate::ingest::authority::MarketKey =
(Arc::from("KALSHI"), Arc::from("sports"), 10, 1);
(Arc::from("KALSHI"), Arc::from("events"), 10, 1);
let hist_key = history::Key {
source_id: 3,
category: "sports".into(),
category: "events".into(),
channel: 10,
instrument_id: 1,
};
Expand All @@ -2070,12 +2070,12 @@ mod tests {
symbol: "NARROWED".into(),
channel: 10,
instrument_id: 1,
category: "sports".into(),
category: "events".into(),
price_exponent: -4,
qty_exponent: -2,
},
);
seed_book(&r, "KALSHI", 3, "NARROWED", "sports", 10, 1);
seed_book(&r, "KALSHI", 3, "NARROWED", "events", 10, 1);
r.cfg.history.lock().unwrap().ingest(
hist_key.clone(),
history::Print {
Expand Down Expand Up @@ -2149,14 +2149,14 @@ mod tests {
vec![sports_row()],
ChannelFilter::parse("edge-kalshi-sports-mbp=10,11").unwrap(),
);
let key10 = ("KALSHI", "sports", FeedKind::MarketByPrice, 34010u16);
let key10 = ("KALSHI", "events", FeedKind::MarketByPrice, 34010u16);
let catalog_key: (Arc<str>, Arc<str>, u8, u32) =
("KALSHI".into(), "sports".into(), 10u8, 1u32);
("KALSHI".into(), "events".into(), 10u8, 1u32);
let book_key: crate::ingest::authority::MarketKey =
(Arc::from("KALSHI"), Arc::from("sports"), 10, 1);
(Arc::from("KALSHI"), Arc::from("events"), 10, 1);
let hist_key = history::Key {
source_id: 3,
category: "sports".into(),
category: "events".into(),
channel: 10,
instrument_id: 1,
};
Expand Down Expand Up @@ -2186,12 +2186,12 @@ mod tests {
symbol: "STILLHERE".into(),
channel: 10,
instrument_id: 1,
category: "sports".into(),
category: "events".into(),
price_exponent: -4,
qty_exponent: -2,
},
);
seed_book(&r, "KALSHI", 3, "STILLHERE", "sports", 10, 1);
seed_book(&r, "KALSHI", 3, "STILLHERE", "events", 10, 1);
r.cfg.history.lock().unwrap().ingest(
hist_key.clone(),
history::Print {
Expand Down Expand Up @@ -2263,9 +2263,9 @@ mod tests {
vec![sports_row()],
ChannelFilter::parse("edge-kalshi-sports-mbp=10,11").unwrap(),
);
let key10 = ("KALSHI", "sports", FeedKind::MarketByPrice, 34010u16);
let key10 = ("KALSHI", "events", FeedKind::MarketByPrice, 34010u16);
let catalog_key: (Arc<str>, Arc<str>, u8, u32) =
("KALSHI".into(), "sports".into(), 10u8, 1u32);
("KALSHI".into(), "events".into(), 10u8, 1u32);

// Tick 1: both channels admitted (spawns real, never-polled receivers).
r.tick().await;
Expand All @@ -2286,7 +2286,7 @@ mod tests {
// Seed the catalog entry the departure is supposed to purge.
r.cfg.instruments.lock().unwrap().insert(
catalog_key.clone(),
test_instrument_in("sports", "KALSHI", 3, "NARROWED", 10, 1),
test_instrument_in("events", "KALSHI", 3, "NARROWED", 10, 1),
);

// Narrow the filter: channel 10 departs. Tick 2 aborts the stand-in and queues it in
Expand All @@ -2302,7 +2302,7 @@ mod tests {
// ordering the finding describes.
r.cfg.instruments.lock().unwrap().insert(
catalog_key.clone(),
test_instrument_in("sports", "KALSHI", 3, "LATE-WRITE", 10, 1),
test_instrument_in("events", "KALSHI", 3, "LATE-WRITE", 10, 1),
);

// Give the runtime a chance to actually process the stand-in's cancellation: it's parked
Expand Down
8 changes: 6 additions & 2 deletions src/ingest/registry.json
Original file line number Diff line number Diff line change
Expand Up @@ -251,18 +251,22 @@
},
{
"venue": "KALSHI",
"category": "sports",
"category": "events",
Comment thread
juan-malbeclabs marked this conversation as resolved.
"code": "edge-kalshi-sports-mbp",
"kind": "MarketByPrice",
"group": "233.84.178.20",
"emit_trades": true,
"arbitration": "Sticky",
"publisher_offset": 100,
"notes": [
"Kalshi sports markets, market-by-price depth. A universe disjoint from the perps rows under the",
"Kalshi event markets, market-by-price depth. A universe disjoint from the perps rows under the",
"same Source ID, which is what `category` expresses: the two never contest each other's tape or",
"book.",
"",
"NAMING. The `category` is `events` - sport is one kind of event market and the universe carries",
"more than sport - while the group `code` stays edge-kalshi-sports-mbp: that name is the ledger's",
"and a subscriber matches `doublezero status` on it, so it is not ours to rename here.",
Comment thread
juan-malbeclabs marked this conversation as resolved.
"",
"CHANNELS. One channel is one league or sport, each an independent state machine with its own reset",
"count, sequence series, manifest seq and snapshot cycle - so one channel is one publisher and one",
"receiver task, with its own processor state and books. Ids are permanent from first publication:",
Expand Down
12 changes: 6 additions & 6 deletions src/ingest/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1144,8 +1144,8 @@ fn leak(s: &str) -> &'static str {
Box::leak(s.to_string().into_boxed_str())
}

/// The sports row's published set of channels, read from the **document** rather than from the
/// expanded rows.
/// The `events` row's published set of channels (group code `edge-kalshi-sports-mbp`), read from the
/// **document** rather than from the expanded rows.
///
/// A test helper, not the authority: reading the document is what lets the port tests assert
/// the `base + id` derivation instead of merely echoing whatever the expander produced.
Expand All @@ -1155,15 +1155,15 @@ pub(crate) fn sports_channel_ids() -> Vec<u8> {
let row = doc
.feeds
.iter()
.find(|f| f.category == "sports")
.expect("built-in document has no sports row");
.find(|f| f.category == "events")
.expect("built-in document has no events row");
match &row.publishers {
Publishers::Derived(d) => published_set_entries(row, &d.channels)
.expect("sports published set")
.expect("events published set")
.into_iter()
.map(|e| e.id)
.collect(),
Publishers::Explicit(_) => panic!("the sports row must carry a derived published set"),
Publishers::Explicit(_) => panic!("the events row must carry a derived published set"),
}
}

Expand Down
19 changes: 9 additions & 10 deletions src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -967,14 +967,13 @@ mod tests {
.map(|f| f.category)
.collect();
mbp.sort_unstable();
// The categories themselves, not a count: a count of 2 would also be satisfied by the same
// universe selected twice, which is the opposite failure and equally wrong.
// The categories themselves, each exactly once. Elections joined perps and sports as a
// third market-by-price universe on this venue, which is the case the dedup has to keep
// getting right rather than a reason to loosen the assertion to a count.
// The categories themselves, each exactly once — not a count, which the same universe
// selected twice would also satisfy, the opposite failure and equally wrong. Elections
// joined perps and events as a third market-by-price universe on this venue, which is the
// case the dedup has to keep getting right rather than a reason to loosen the assertion.
assert_eq!(
mbp,
vec!["elections", "perps", "sports"],
vec!["elections", "events", "perps"],
"every universe must be selected, each once"
);
}
Expand Down Expand Up @@ -1092,8 +1091,8 @@ mod tests {
registry();
let sports = feeds::feeds()
.iter()
.find(|f| f.category == "sports")
.expect("the built-in registry has a sports row");
.find(|f| f.category == "events")
.expect("the built-in registry has an events row");
let chan10_port = sports
.publishers
.iter()
Expand Down Expand Up @@ -1129,8 +1128,8 @@ mod tests {
registry();
let sports = feeds::feeds()
.iter()
.find(|f| f.category == "sports")
.expect("the built-in registry has a sports row");
.find(|f| f.category == "events")
.expect("the built-in registry has an events row");
let chan10_port = sports
.publishers
.iter()
Expand Down
7 changes: 5 additions & 2 deletions src/sinks/admin.rs
Original file line number Diff line number Diff line change
Expand Up @@ -429,11 +429,14 @@ mod tests {
/// ([`ChannelFilter::parse`]'s own docs), so a fixture built from an ad hoc `Feed` (an unknown
/// code) would make every `POST` in this module 400 regardless of what this handler does; using
/// the real feed is what lets a valid spec actually apply.
///
/// Named for the group `code` (`edge-kalshi-sports-mbp`), which is the ledger's and unchanged,
/// not for the `category` — that is `events`, since the row carries more than sport.
fn sports_row() -> Feed {
*crate::ingest::feeds::feeds()
.iter()
.find(|f| f.category == "sports")
.expect("the built-in registry has a sports row")
.find(|f| f.category == "events")
.expect("the built-in registry has an events row")
}

/// The sports row narrowed to a single publisher (`channel`'s), mirroring what
Expand Down
Loading
Loading