diff --git a/CHANGELOG.md b/CHANGELOG.md index cd813ef..4e36959 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ every release's upgrade notes. ### Added +- Expose block HTTP attempts by topology target and published reader snapshot + sizes to diagnose startup distribution without per-connection traffic capture. + - Configure the preferred owner with `-root-fanout` (default 1), independently of descendant `-fanout` (default 2), reducing normal first-hop replication (#100, ADR-0011). Hedging, root fallback and snapshot churn remain unrestricted @@ -30,6 +33,10 @@ every release's upgrade notes. ### Fixed +- Publish new tracker readers immediately and refresh active client snapshots + with bounded, demand-driven backoff (#102, ADR-0012). Concurrent JOINs share + one request; departures remain tick-batched and no background polling is added. + - Coalesce normal block misses across relay hops (#97, ADR-0010). Include eligible self in local reader trees and preserve strictly earlier parent/ancestor edges. Terminal supply remains independent of waiting relays; cold requests crossing diff --git a/contrib/bench/tracker-startup.py b/contrib/bench/tracker-startup.py new file mode 100644 index 0000000..b47d3a2 --- /dev/null +++ b/contrib/bench/tracker-startup.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Compare startup publication/refresh policies using temporary Go overlays.""" +import argparse +import json +import os +from pathlib import Path +import subprocess +import tempfile + +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument('--count', type=int, default=3) +parser.add_argument('--policy', choices=['all', 'frozen', 'normal', 'once', 'backoff'], default='all') +parser.add_argument('--scenario', choices=['burst', 'staggered', 'reverse', 'slow-growth']) +args = parser.parse_args() +if args.count < 1: + parser.error('--count must be positive') +repo = Path(__file__).resolve().parents[2] +reader_path = repo / 'internal/engine/readers.go' +tracker_path = repo / 'internal/tracker/tracker.go' +readers = reader_path.read_text() +start = readers.index('func readerRefreshDelay(') +end = readers.index('\n// trackerAddr', start) +normal = readers[:start] + '''func readerRefreshDelay(previous, ceiling time.Duration) time.Duration { + return ceiling +} +''' + readers[end:] +once = readers[:start] + '''func readerRefreshDelay(previous, ceiling time.Duration) time.Duration { + if previous <= 0 { return min(initialReaderRefresh, ceiling) } + return ceiling +} +''' + readers[end:] +tracker = tracker_path.read_text() +start = tracker.index('\t// Growth must be visible in this JOIN response.') +end = tracker.index('\n\treturn JoinResponse{', start) +frozen = tracker[:start] + tracker[end:] +policies = ['frozen', 'normal', 'once', 'backoff'] if args.policy == 'all' else [args.policy] +with tempfile.TemporaryDirectory(prefix='dart-startup-overlay-') as tmp: + tmp = Path(tmp) + for policy in policies: + replacements = {} + if policy != 'backoff': + candidate = tmp / 'readers.go' + candidate.write_text(once if policy == 'once' else normal) + replacements[str(reader_path)] = str(candidate) + if policy == 'frozen': + candidate = tmp / 'tracker.go' + candidate.write_text(frozen) + replacements[str(tracker_path)] = str(candidate) + overlay = tmp / 'overlay.json' + overlay.write_text(json.dumps({'Replace': replacements})) + env = dict(os.environ, DART_TRACKER_STARTUP_EXPERIMENT='1', DART_REFRESH_POLICY=policy) + env.setdefault('GOMAXPROCS', '2') + pattern = '^TestTrackerStartupExperiment$' + if args.scenario: + pattern += '/^' + args.scenario + '$' + subprocess.run(['go', 'test', '-overlay', str(overlay), './internal/engine', + '-run', pattern, '-count', str(args.count), '-timeout', '5m', '-v'], + cwd=repo, env=env, check=True) diff --git a/docs/README.md b/docs/README.md index 20853bd..1ea4c94 100644 --- a/docs/README.md +++ b/docs/README.md @@ -15,7 +15,7 @@ This directory holds DART's **official documentation** (tracked in git). It targ | [fetch.md](./fetch.md) | `internal/fetch` — origin HTTP range fetch + singleflight coalescing | Implemented, tests passing | | [engine.md](./engine.md) | `internal/engine` — read-through orchestration (cache/peer/origin), whole-block relay, bounded prefetch, hedging | Implemented, tests passing | | [peer.md](./peer.md) | `internal/peer` — node-to-node block transport (server + pooled client) | Implemented, tests passing | -| [tracker.md](./tracker.md) | `internal/tracker` — per-file active reader set (leases, tick freeze) | Implemented, tests passing | +| [tracker.md](./tracker.md) | `internal/tracker` — per-file active reader set (leases, immediate growth, batched departures) | Implemented, tests passing | | [registry.md](./registry.md) | `internal/registry` — container registry pull-through mirror (OverlayBD/containerd path) | Implemented, tests passing | | [observability.md](./observability.md) | `internal/metrics` + `internal/admin` — Prometheus exporter, admin endpoints | Implemented, tests passing | | [node.md](./node.md) | `internal/node` — node assembly: flags, store/engine wiring, discovery-scheme dispatch | Implemented, tests passing | diff --git a/docs/adr/0012-publish-reader-growth-immediately.md b/docs/adr/0012-publish-reader-growth-immediately.md new file mode 100644 index 0000000..66cb1ed --- /dev/null +++ b/docs/adr/0012-publish-reader-growth-immediately.md @@ -0,0 +1,70 @@ +# ADR-0012: Publish reader growth immediately and refresh on bounded demand + +- Status: accepted +- Date: 2026-11-09 +- Supersedes: none +- Binds: docs/tracker.md publication, integration and Stability Contract; docs/engine.md reader refresh and Stability Contract; docs/observability.md topology metrics; docs/dart.md tracker-tick + +## Context + +Issue [#102](https://github.com/data-accelerator/dart/issues/102) identifies two +startup delays: freezing the first JOIN's singleton reader set until a tracker +tick, then retaining that response in client caches for half a reader lease. +Local self completion keeps routing acyclic but cannot discover other readers. +Even with root fanout one, many different singleton snapshots can send all +readers directly to the root. Normal reader forwarding already follows strictly +earlier HRW ranks under common member weights; correctness does not require +synchronized reader snapshots or frozen topology. + +## Decision + +Publish newly joined readers before replying, batch departures on ticks, and +refresh client snapshots only on routing demand with shared per-object/view +JOINs and time-based doubling from 50 ms to half the granted lease. + +## Consequences + +Renewing an already published reader changes no publication epoch unless a due +removal sweep changes the set. New JOINs do not postpone removal processing; +last-reader deletion and idle eviction retain their immediate/lazy semantics. +Snapshots remain sorted copies, and publication epochs remain local to one +tracker file entry. JSON and peer wire framing do not change. + +An active client's successful refresh intervals double independently of reader +list changes. Short grants cap even the initial interval; missing grants retain +a two-second ceiling. Expired leases or a different tracker membership view +restart the schedule. There are no background refresh workers. Waiters may +cancel independently; a canceled leader can be replaced without publishing its +late result. Failed JOINs are shared but not cached. A response started before +a more recently installed view cannot overwrite that view. + +A synthetic warm-root experiment supports repeated early refresh for gradual +arrivals, while a single early refresh is cheaper for completed bursts. Fifty +milliseconds is a starting engineering choice, not an optimal cluster constant. +Late arrivals after the schedule reaches its ceiling still wait for the next +routing lookup. Refresh traffic scales with active node/object pairs; this +change makes no thousand-reader control-plane scalability claim. + +Expose low-cardinality counters for dispatched root/parent/ancestor block +attempts and a histogram of reader counts in successful JOIN replies. Exclude +pre-dispatch rejections, local hits and coalesced JOIN waiters. Attempts are not +wire-byte counters; raw reader counts are not filtered tree sizes. + +The preferred owner, local self completion, strict earlier-rank forwarding, +normal cross-hop coalescing, terminal supply flights and ordering-epoch boundary +remain unchanged. Mixed versions retain read correctness but obtain only +partial startup performance improvement; coordinated upgrade is recommended. + +## Alternatives considered + +- Tick-frozen growth and normal half-lease refresh preserve the initial root + bottleneck despite a healthy, warm root and available reader peers. +- A single early refresh has lower control-plane cost for bursts but can finish + before later readers join; simple doubling also covers gradual startup. +- Repeated fixed fast polls continue spending requests after startup. Background + refresh also spends requests on completed reads. +- Notifications, gossip convergence, reader-change-triggered resets and global + synchronization add state or RPC dependencies that ordered forwarding does + not need. This remains a trusted-cluster design. +- Per-object or per-peer metric labels have unbounded operational cardinality; + aggregate topology classes and snapshot-size observations suffice here. diff --git a/docs/adr/README.md b/docs/adr/README.md index 8f6b877..d100016 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -135,3 +135,4 @@ acceptable in current law — it is a pointer, not a narrative. | [0009](./0009-root-eligible-peer-roles.md) | Anchor distribution at role-eligible owners | accepted | | [0010](./0010-coalesce-directed-block-loads.md) | Coalesce directed normal block loads across hops | accepted | | [0011](./0011-separate-root-fanout.md) | Configure root fanout independently of reader branching | accepted | +| [0012](./0012-publish-reader-growth-immediately.md) | Publish reader growth immediately and refresh on bounded demand | accepted | diff --git a/docs/dart.md b/docs/dart.md index 4e95a94..901e041 100644 --- a/docs/dart.md +++ b/docs/dart.md @@ -55,7 +55,7 @@ dart [flags] | `-root-fanout` | `1` | preferred root normal branching factor (≥1); not a limit on hedge/fallback transfers | | `-admin` | `:9100` | admin/metrics listen address; empty disables | | `-reader-tree` | `true` | build the distribution tree over the active reader set (per-file tracker) | -| `-tracker-tick` | `3s` | how often a tracker republishes a reader set | +| `-tracker-tick` | `3s` | removal/expiry batching interval; new readers publish immediately | | `-replicas` | `1` | HRW candidates that authoritatively hold a chunk (the owned budget) | | `-owned-fraction` | `0.8` | owned share in [0,1]; 0 is borrowed-only, 1 owned-only; invalid/non-finite values fail startup | | `-mem-size` | `256 MiB` | Hybrid hot-set size; ignored in memory/disk modes; below one block disables the Hybrid front tier | diff --git a/docs/engine.md b/docs/engine.md index d2e2182..38d12bc 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -32,14 +32,23 @@ never become descendants; non-root peers never contact origin. Without P2P, ordinary nodes operate as single-node caches; node assembly requires P2P for restricted roles. -Reader-set lookups are **cached per object for half the granted tracker lease** -(and the cache doubles as the lease renewal: a hit re-serves the frozen set, a -miss re-JOINs, which refreshes this node's own lease). Caching for half the -lease guarantees renewal always lands before the lease lapses; a tracker that -grants nothing falls back to a 2 s period. Expired entries are swept as the -cache grows past a few hundred objects, so the map tracks objects currently -being read and stays small. When the tracker is unreachable the engine falls -back to the preferred owner for that lookup (the reader set is soft state). +Reader-set lookups are cached per object and tracker membership view (ADR-0012). +On continued routing demand, successful refreshes use 50 ms, 100 ms, 200 ms, ..., +capped at half the granted tracker lease. A short lease caps even the first +interval; a missing grant uses a 2 s ceiling. Each miss re-JOINs to renew the +lease. The schedule restarts after lease expiry or tracker-view change, never +because the reader list changed. There are no background polls or refresh RPCs +on an ordinary cache hit. Concurrent JOINs share one inline caller; waiters can +cancel independently and retry after leader cancellation. Failed refreshes are +not cached, and late responses cannot replace a more recently installed view. +Entries past their lease are swept as the map grows beyond 256 objects. When +the tracker is unreachable that lookup routes to the preferred owner. +A fresh cache hit reuses tracker selection for the identical immutable membership +view, avoiding full-cluster ranking and allocations. A replacement view is +re-evaluated even when its ordering epoch is unchanged, because addresses and +liveness are not part of that epoch. +See [tracker measurements](./tracker.md#startup-measurement-and-tradeoffs) for +startup traffic and the additional control-plane request cost. ## 2. Concepts @@ -579,6 +588,10 @@ block sources. A supply root records its proxied response bytes as `origin_in`. ## 6. Determinism / Stability Contract +Reader JOIN growth is immediate; client refresh uses demand-driven bounded +backoff and shared per-object/view calls (ADR-0012, [tracker](./tracker.md)). +This does not alter HRW order, normal cross-hop flights or terminal supply. + - ADR-0011 defaults the preferred root to one child and descendants to two. Explicit equal degrees preserve the uniform tree. Changing degrees preserves the strict HRW dependency order; no flight key or membership epoch change is @@ -718,7 +731,8 @@ go test ./internal/engine/ -cover -count=1 | `TestBlockDoesNotCacheRangeIgnored` | defense in depth: bytes sliced from a Range-ignored 200 are served, never cached | | `TestHedgeWinMetricsRequireFiredHedge` | no hedge fired → no hedge-win counters (the backup_won/fired comparison stays meaningful) | | `TestMissesDoNotFeedLatencyEstimator` | fast 404 misses never arm the hedge delay | -| `TestReaderSetCacheFollowsGrantedLease` | the reader-set cache renews at half the granted lease, never lapsing mid-read | +| `TestReaderSetCacheFollowsGrantedLease` | a short granted lease bounds the reader-set cache renewal period | +| `TestPeerFetchTargetMetrics` | real root, parent and ancestor attempts match metrics, including breaker-promoted ancestors | | `TestRelayParentErrorFallsBackToOrigin` | an open-circuit/errored parent is skipped to origin, not propagated as a 500 | | `TestEmptyObjectGetServes200` | empty object: 200 empty on both paths; a Range request stays 416 | | `TestParseRangeSuffixEmptyObject` | a suffix range on an empty object is unsatisfiable (no inverted `[0,-1]` interval); non-empty suffix behavior unchanged | diff --git a/docs/observability.md b/docs/observability.md index e5239a4..7ed363e 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -77,6 +77,8 @@ and safe for concurrent use, since it runs on the scrape path. | `dart_relay_total` | counter | `result=served\|declined` | relay requests handled for peers | | `dart_block_fetch_seconds` | histogram | `source=peer\|origin` | block-path fetch latency: cache fills and peer reads only — Size probes and the uncached passthrough path are **not** observed here (`dart_passthrough_total` counts the latter). A coalesced fetch still observes its latency (the caller genuinely waited) even though its bytes are not double-counted in `dart_bytes_total` | | `dart_hedge_total` | counter | `event=fired\|primary_won\|backup_won` | tail-latency hedges: duplicates launched, and which contender served the block | +| `dart_peer_fetch_total` | counter | `target=root,parent,ancestor` | dispatched block HTTP attempts; root includes fallback supply roots; parent means the direct tree parent, ancestor a higher non-root ancestor | +| `dart_reader_snapshot_size` | histogram | — | reader count in successful JOIN replies; one observation per completed shared refresh, not per waiter or cache hit | | `dart_peer_failover_total` | counter | — | definite peer failures escalated to the next ancestor (not rate limited, unlike hedges) | | `dart_passthrough_total` | counter | `reason=range_unsupported` | requests proxied without block caching (direct for root-capable nodes, delegated for non-root nodes; see docs/engine.md §3.9) | | `dart_peer_circuits_open` | gauge | — | peers whose circuit is currently open | @@ -109,6 +111,31 @@ just briefly slow. The store and breaker metrics are wired by `engine.RegisterStoreMetrics(reg, store)` and `engine.RegisterPeerMetrics(reg, breaker)`; `cmd/dart` calls both. +### Inspecting reader-tree startup + +`dart_peer_fetch_total` classifies from the requesting node's tree snapshot. +A root is counted as `root` even when it is the direct parent; a breaker-skipped +parent does not turn the promoted ancestor into `parent`. Buffer admission +rejections, circuit rejection and local hits do not count. Failed or canceled +after-dispatch attempts and hedges do count; internal HTTP retries do not add +another observation. These are attempts, not successful blocks or wire bytes. + +```promql +# Root share of dispatched block attempts across the selected instances. +sum(rate(dart_peer_fetch_total{target="root"}[1m])) + / sum(rate(dart_peer_fetch_total[1m])) +# Fraction of observed JOIN responses with at most one published reader. +sum(rate(dart_reader_snapshot_size_bucket{le="1"}[1m])) + / sum(rate(dart_reader_snapshot_size_count[1m])) +``` + +The histogram measures raw published IDs before membership/role filtering and +local-self completion, not the size of every currently cached client view. +It also observes local-registry JOINs. No file, URL or peer-ID label is emitted. +Compare root share, singleton observations and existing byte/hedge/failover +metrics over an active interval. These signals reveal likely startup collapse; +they cannot prove a globally consistent tree or measure per-target bandwidth. + ## 4. `internal/admin` endpoints | Method | Path | Response | diff --git a/docs/peer.md b/docs/peer.md index 4f88296..c07896c 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -147,6 +147,8 @@ type LeaseSource func(context.Context, BlockRequest) (store.Lease, bool, error) // Server.Borrow LeaseSource takes precedence over Server.Src. func (c *Client) GetInto(ctx context.Context, addr string, req BlockRequest, dst []byte, validate func(context.Context, int64) error) ([]byte, bool, error) +func (c *Client) GetIntoObserved(ctx context.Context, addr string, req BlockRequest, + dst []byte, validate func(context.Context, int64) error, onStart func()) ([]byte, bool, error) ``` `LeaseSource` transfers immutable pinned bytes to the server, including any lease @@ -163,6 +165,14 @@ It reads at most len(dst)+1 bytes, without a growing payload allocation. The known-length GetValidated compatibility API allocates one expected-size payload buffer. Read-only sharing and cache publication belong to the engine (ADR-0010). +`GetIntoObserved` adds an inline callback immediately before HTTP dispatch. It +must return promptly and runs outside breaker locks. Breaker rejection, invalid +request construction and already-canceled contexts do not notify. Dispatch can +still fail before bytes reach the network; net/http internal retries are not +separate observations. The client's own deadline expiring before dispatch still +records a breaker failure, as it does during transport; it is distinct from +caller cancellation. This is attempt instrumentation, not a byte counter. + ### 3.4 Validated whole-block reads: `Client.GetValidated` ```go @@ -435,6 +445,11 @@ go test ./internal/peer/ -race -count=1 ### Test list (property each guards) +`TestGetIntoObservedDispatch` verifies pre-dispatch observation on real HTTP +success/miss/error and exclusion of breaker/setup/cancellation rejections. +`TestGetIntoObservedClientTimeoutBeforeDispatch` verifies that an undispatched +client-owned timeout cannot falsely recover a half-open circuit. + | Test | Property guarded | |---|---| | `TestServerClientRoundtrip` | store → server → client returns exact block bytes | diff --git a/docs/tracker.md b/docs/tracker.md index f8d8b45..04496cf 100644 --- a/docs/tracker.md +++ b/docs/tracker.md @@ -1,7 +1,8 @@ # `internal/tracker` The per-file **active reader set** `S`: which nodes are currently reading a given -object. The distribution tree is built over `S` instead of over all members. +object. Below the fixed chunk owner, the distribution tree is built over `S` +instead of over all members. - Source: `internal/tracker/tracker.go`, `internal/tracker/client.go` - Tests: `internal/tracker/tracker_test.go`, `internal/tracker/eviction_test.go` @@ -20,8 +21,9 @@ Three properties make this safe and cheap: - **Leases** — a reader JOINs with a TTL and refreshes; when it stops reading the lease lapses and it drops out of `S`, shrinking the tree. `S` is soft state, so a tracker restart self-heals as readers renew. -- **Tick freeze** — the set is recomputed only on a fixed tick, so topology (and - `epochS`) is stable between ticks and peer connections do not churn. +- **Immediate growth, batched departures** — JOIN publishes new readers before + replying. Renewals leave the set and `epochS` unchanged; removals and expiry + are collected on a tick (ADR-0012). - **Control plane only** — small JSON messages; no block data flows through a tracker. @@ -39,8 +41,8 @@ package therefore holds no placement logic. |---|---| | file key | `chunk.ChunkKey(ns, objectID, -1)` — a sentinel index so it never collides with a real chunk's placement key. | | lease | A reader's membership deadline; refreshed by re-JOIN. | -| frozen set | The published reader list, recomputed on a tick and sorted by node ID. | -| `epochS` | Bumped only when the frozen set actually changes. | +| published set | The sorted reader list: grows during JOIN, shrinks on a tick. | +| `epochS` | Bumped only when the published set actually changes. | ## 3. Public API @@ -54,7 +56,7 @@ func (r *Registry) Readers(file string) ([]string, uint64) func (r *Registry) Files() int ``` -- `Join` records or refreshes interest and returns the currently **frozen** set +- `Join` records or refreshes interest and returns the currently **published** set (`JoinResponse{EpochS, Readers, TTLMs}`). `ttl <= 0` uses the registry default; `ttl > MaxLeaseTTL` (10m) is **clamped** to `MaxLeaseTTL`, and the granted `TTLMs` always reflects the clamp. Leases are meant to be refreshed on a @@ -63,15 +65,17 @@ func (r *Registry) Files() int On the HTTP wire, `ttlMs` values too large for a `time.Duration` are clamped *before* conversion, so they can never overflow-wrap into a silently wrong lease. -- The first reader of a file is published immediately (it would be useless to - make the first reader wait a whole tick); subsequent changes wait for the tick. -- `Leave` drops a lease immediately; the frozen set updates on the next tick. +- Every new reader is published before its JOIN returns. A renewal of an already + published reader does not rebuild the set or change `epochS` unless a due tick + also removes other readers. Growth does not postpone the removal tick. +- `Leave` drops a lease immediately; the published set updates on the next tick. Dropping the last reader forgets the file entirely. - A lease expires at its TTL deadline regardless of further Joins — expiry is not "until the next Join". Enforcement is **lazy**: the sweeper runs on registry activity (amortized to at most once per tick; there is no background goroutine), so an expired lease in an untouched registry lingers - until the next call — harmless, because only live readers shape the tree. + until the next call. A published departure may linger until the next tick; + stale parents are handled by normal forwarding/fallback. - `Readers` returns a **copy**; `Files` is a diagnostic count. - **Idle eviction**: a file with no live leases (readers vanished without a `Leave`) and no `Join`/`Readers` activity for `IdleGrace` is forgotten, so a @@ -107,21 +111,18 @@ capped at 64 KiB. ## 4. Invariants & Guarantees -- **Tick freeze**: the published reader set changes only at tick boundaries - (recomputation is lazy — on activity, at most once per tick; there is no - background goroutine). Between ticks the topology and `epochS` are stable, - so TCP connections to readers do not churn. -- **`epochS` bumps only when the frozen set actually changes** — joins that - refresh an existing lease, and leaves of absent readers, never bump it. -- **Deterministic wire form**: the frozen set is sorted by node ID; every +- **Immediate growth**: a successful JOIN includes its reader in the published + set, without waiting for a tick. Renewal of an existing reader changes neither + the set nor `epochS` unless a due removal sweep changes it. +- **Batched departures**: removal/expiry recomputation is lazy, at most once per + tick. No background goroutine runs, and new readers do not delay that sweep. + `Leave` deletes the lease immediately; deleting the last lease forgets the + file, so `Readers` reports `(nil, 0)` without waiting for a tick. An idle entry + is forgotten after `IdleGrace`. +- **`epochS` changes only with the published set**. It is local to one file entry + on one tracker, not comparable across trackers or entry recreation. +- **Deterministic wire form**: the published set is sorted by node ID; every observer of the same registry state reads the same epoch and reader list. -- **Membership by liveness only**: a reader is in the set iff its lease is - unexpired at freeze time. `Leave` deletes the lease immediately; for a file - with remaining readers the published set follows at the next freeze, but - removing the **last** lease deletes the whole file entry, so `Readers` then - reports `(nil, 0)` without waiting for a tick (the tick-freeze guarantee - covers joins and lease expiries, not this deletion edge). An idle entry is - forgotten after `IdleGrace`. - **No collision with placement keys**: the file key uses the sentinel chunk index -1, which no real chunk's `ChunkKey` input can carry. - **Client-supplied TTLs are clamped** to the configured lease range; @@ -134,8 +135,14 @@ capped at 64 KiB. uses all eligible Ready readers below the owner. The engine then: 1. computes the file key and the HRW tracker for it; -2. JOINs (locally if it *is* the tracker, else over HTTP) and caches the reader - set for ~2s, so a read does not make a tracker call per block; +2. JOINs locally or over HTTP; concurrent refreshes for the same object and + tracker membership view share one request. On continued routing demand, the + cache interval starts at 50 ms and doubles after each successful refresh, + capped at half the granted lease (3 s for the default 6 s lease). There is no + background polling; completed reads cause no further JOINs. A short grant + caps even the first interval. Missing grants use a 2 s ceiling. The schedule + resets after lease expiry or a changed tracker membership view, not after a + changed reader list or failed JOIN; 3. anchors the tree at the root-eligible chunk owner, appends ranked eligible readers union eligible self, deduplicates them (including the owner), and takes its parent from that sequence. This local completion does not mutate the @@ -153,13 +160,19 @@ do not move ownership; loss of all usable roots can still fail a cache miss. - `Registry` is safe for concurrent use (single mutex); `Readers`/`Join` return copies, never internal slices. Verified with `-race`. - `Client` is safe for concurrent use. -- The engine's reader-set cache is mutex-guarded and holds only IDs. +- The engine's reader-set cache is mutex-guarded and publishes immutable ID slices. + Waiting callers can cancel independently. A canceled refresh leader permits a + live waiter to retry; an abandoned or older response cannot overwrite a newer + installed entry. A failed JOIN is shared by its waiters and is not cached. + Fresh hits reuse tracker selection for the same immutable membership view; a + new view rechecks address/liveness even if its ordering epoch is unchanged. ## 7. Stability Contract - The JSON shapes and paths (`/tracker/v1/join`, `/leave`) are the tracker wire protocol; changing them is a protocol change. -- The frozen set is **sorted**, so every node derives an identical tree from the +- Publish new readers during JOIN and batch departures on ticks (ADR-0012). +- The published set is **sorted**, so every node derives an identical tree from the same set — this is load-bearing and must not regress. - `epochS` must change only when the set changes (readers rely on it to detect topology changes cheaply). @@ -180,7 +193,8 @@ go test ./internal/tracker/ -race -count=1 | Test | Property guarded | |---|---| | `TestJoinPublishesFirstReaderImmediately` | the first reader is published without waiting a tick | -| `TestTickFreeze` | a join within a tick does not change the published set or epoch; the next tick does | +| `TestJoinPublishesGrowthImmediately` | every JOIN publishes growth, keeps sorted copied snapshots and leaves renewal epochs unchanged | +| `TestDeparturesWaitForTick` | joins do not postpone removal batching; rejoining a removed reader publishes immediately | | `TestLeaseExpiry` | a reader that stops renewing drops out | | `TestJoinClampsLeaseTTL` | a 24h lease request is granted clamped to `MaxLeaseTTL` and actually lapses then | | `TestJoinTTLMsOverflowIsClamped` | overflowing `ttlMs` values clamp before duration conversion (no wrap) | @@ -204,17 +218,84 @@ Engine-side (in `internal/engine`): | `TestTreeNodesPrefersReaderSet` | tree anchors at the owner and uses eligible active readers below it | | `TestTreeNodesWithoutReaderSnapshot` | tracking disabled uses eligible readers; a tracked sole reader retains the owner | | `TestReaderSetTreeEndToEnd` | 3 nodes sharing one tracker all read correct bytes | +| `TestTrackerStartupPublishesBeforeTick` | real HTTP payload counts prove reader P2P before the first tick | +| `TestTrackerStartupExperiment` | opt-in 1 MiB Range latency, root/reader payload and JOIN cost comparison | +| `TestReaderRefreshCoalescesAndCancelsWaiters` | one refresh per object/view, independent waiter cancellation, no background polling, one histogram observation | +| `TestReaderRefreshCanceledLeaderRetries` | a live waiter replaces a canceled leader without retaining a flight | +| `TestReaderRefreshOldTrackerCannotOverwriteNew` | a late response cannot replace a newer tracker view; tracker epochs are not globally ordered | +| `TestReaderRefreshLeaseBounds` | small, missing and overflowing grants preserve cache lease bounds | +| `TestReaderRefreshBackoffAndLeaseReset` | deterministic doubling, half-lease cap, and restart after idle lease expiry | +| `TestReaderRefreshFailureDoesNotCache` | failed JOIN is shared and a later demand can retry | +| `TestReaderSnapshotCacheHitDoesNotAllocate` | stable-view hits avoid allocation; canceled lookups return no cached readers | +| `TestReaderSnapshotCacheTracksAddressAndLiveness` | address and liveness changes invalidate selection despite unchanged epoch | ## 9. Limitations & TODO - **Tracker liveness**: a tracker that dies is replaced by HRW on the next membership change, and `S` rebuilds from renewals — but in-flight reads during the gap route directly to the owner. -- **No epochS-driven invalidation**: the engine caches the reader set for a fixed - ~2s rather than reacting to an `epochS` change pushed by the tracker. +- **No pushed invalidation**: the client does not learn about later JOINs until + its next due routing lookup. Late arrivals after backoff reaches its ceiling + can wait up to that ceiling. A tracker restart at an unchanged membership view + likewise rebuilds on normal renewals; there is no event-driven acceleration. - **No append-only mode**: the design's option to add new readers as leaves and - only rebalance on a tick boundary is not implemented; every tick may reshape the - tree tail. + only rebalance on a tick boundary is not implemented; growth and departure + publication can reshape the tree. Common-reader HRW order remains unchanged. - **Memory**: bounded by **idle eviction** (§3.1): a file with no live leases and no activity for `IdleGrace` is forgotten. There is deliberately no hard file cap — a hard cap would evict genuinely active files under load. + +### Startup measurement and tradeoffs + +`TestTrackerStartupPublishesBeforeTick` uses a warm root, eight cold readers and +real HTTP traffic with tracker time held before the first removal tick. Eight +4 MiB reads deliver 8 MiB from the root and 24 MiB from readers, with eight JOINs. +The same test against tick-frozen publication delivered all 32 MiB from the root. +Anchor: `internal/engine/tracker_startup_test.go::TestTrackerStartupPublishesBeforeTick`. + +The opt-in experiment reads 32 MiB per reader as sequential 1 MiB Ranges. It +serializes root service at 40 ms per 4 MiB to expose a root bottleneck; hedging +is off, prefetch window is one, and all bytes are checked. These are synthetic +loopback results, not a NIC throughput benchmark or a universal latency claim. +Median of three runs, eight readers, 256 MiB total client payload: + +| Arrival pattern / refresh | Root MiB | JOINs | Aggregate Gbps | Range p95 / p99 ms | +|---|---:|---:|---:|---:| +| 1 ms burst / frozen publication + half-lease | 256 | 8 | 0.77 | 347 / 349 | +| 1 ms burst / immediate publication + half-lease | 64 | 8 | 3.04 | 89 / 104 | +| 1 ms burst / one early refresh | 36 | 16 | 4.46 | 58 / 101 | +| 1 ms burst / doubling refresh | 36 | 26 | 4.45 | 59 / 103 | +| 120 ms reverse-HRW arrivals / half-lease | 256 | 8 | 0.77 | 301 / 306 | +| 120 ms reverse-HRW arrivals / one early refresh | 176 | 16 | 1.11 | 185 / 255 | +| 120 ms reverse-HRW arrivals / doubling refresh | 136 | 32 | 1.43 | 182 / 287 | + +Doubling helps readers joining after the first early refresh, at the cost of +additional requests. It does not uniformly improve p99: topology changes can +add a relay wait. In the slow-growth comparison, JOIN response JSON totals were +700 / 1433 / 3108 bytes; summed tracker-handler wall times were about +0.54 / 0.89 / 1.46 ms. Handler wall time is not CPU usage. This small experiment +does not establish tracker scalability at thousands of readers or objects. +Without the synthetic root limit, short reads often finish before 50 ms and +produce no additional refresh. The 50 ms start is an initial engineering choice, +not a measured optimum. With a 6 s grant, earliest refresh times are approximately +0, 50, 150, 350, 750, 1550, 3150, 6150 ms, then every 3 s while demanded. + +Reproduce all four policies using temporary Go overlays (no source mutation): + +```bash +python3 contrib/bench/tracker-startup.py --count 3 +# Or only the production policy: +DART_TRACKER_STARTUP_EXPERIMENT=1 go test ./internal/engine \ + -run '^TestTrackerStartupExperiment$' -count=3 -v +``` + +Anchor: `internal/engine/tracker_experiment_test.go::TestTrackerStartupExperiment`. + +`BenchmarkReaderSnapshotCacheHit` isolates the stable-view cache-hit path at +32, 256 and 4096 members. It reports time and allocations separately from the +network experiment; a stable hit requires zero allocations. View replacement +may allocate once to remember an unchanged tracker selection. + +```bash +go test ./internal/engine -run '^$' -bench '^BenchmarkReaderSnapshotCacheHit$' -benchmem +``` diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 58196b1..dbe7877 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -149,6 +149,7 @@ type Engine struct { trackerReg *tracker.Registry rsMu sync.Mutex rs map[string]*readerSet + rsCalls map[readerFlightKey]*readerFlight mu sync.Mutex sizes map[string]sizeMeta // origin endpoint -> probe result diff --git a/internal/engine/metrics.go b/internal/engine/metrics.go index e7ca11f..f8bf9ce 100644 --- a/internal/engine/metrics.go +++ b/internal/engine/metrics.go @@ -36,7 +36,9 @@ type Metrics struct { // Failover: a definite peer failure escalated to the next ancestor. Counted // separately from hedging because it is reactive rather than speculative and // is deliberately not rate limited. - failover *metrics.Counter + failover *metrics.Counter + peerFetchRoot, peerFetchParent, peerFetchAncestor *metrics.Counter + readerSnapshotSize *metrics.Histogram // Passthrough: requests proxied verbatim because the origin ignores Range. passthrough *metrics.Counter @@ -53,11 +55,15 @@ var blockLatencyBounds = []float64{0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10} // NewMetrics registers the engine's metrics on r and returns the handle. func NewMetrics(r *metrics.Registry) *Metrics { return &Metrics{ - temporaryBuffers: r.NewCounter("dart_temporary_buffers_total", "temporary block receive buffers"), - temporaryBytes: r.NewCounter("dart_temporary_buffer_bytes_total", "bytes allocated for temporary block receive buffers"), - receiveLoads: r.NewGauge("dart_receive_buffers", "live temporary block receive buffers including retained results"), - receiveBytes: r.NewGauge("dart_receive_bytes", "live temporary block receive payload bytes including retained results"), - receiveRejected: r.NewCounter("dart_receive_budget_rejected_total", "temporary receive admissions declined, including skipped speculation"), + peerFetchRoot: r.NewCounter("dart_peer_fetch_total", "peer block HTTP attempts dispatched by topology target", metrics.LabelPair{Name: "target", Value: "root"}), + peerFetchParent: r.NewCounter("dart_peer_fetch_total", "peer block HTTP attempts dispatched by topology target", metrics.LabelPair{Name: "target", Value: "parent"}), + peerFetchAncestor: r.NewCounter("dart_peer_fetch_total", "peer block HTTP attempts dispatched by topology target", metrics.LabelPair{Name: "target", Value: "ancestor"}), + readerSnapshotSize: r.NewHistogram("dart_reader_snapshot_size", "published reader count in successful tracker JOIN responses, excluding cache hits and coalesced waiters", []float64{0, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024}), + temporaryBuffers: r.NewCounter("dart_temporary_buffers_total", "temporary block receive buffers"), + temporaryBytes: r.NewCounter("dart_temporary_buffer_bytes_total", "bytes allocated for temporary block receive buffers"), + receiveLoads: r.NewGauge("dart_receive_buffers", "live temporary block receive buffers including retained results"), + receiveBytes: r.NewGauge("dart_receive_bytes", "live temporary block receive payload bytes including retained results"), + receiveRejected: r.NewCounter("dart_receive_budget_rejected_total", "temporary receive admissions declined, including skipped speculation"), cacheHits: r.NewCounter("dart_block_source_total", "blocks served, by source", metrics.LabelPair{Name: "source", Value: "cache"}), peerHits: r.NewCounter("dart_block_source_total", @@ -100,6 +106,26 @@ func NewMetrics(r *metrics.Registry) *Metrics { } } +func (m *Metrics) recordPeerFetch(root, parent bool) { + if m == nil { + return + } + switch { + case root: + m.peerFetchRoot.Inc() + case parent: + m.peerFetchParent.Inc() + default: + m.peerFetchAncestor.Inc() + } +} + +func (m *Metrics) recordReaderSnapshot(n int) { + if m != nil { + m.readerSnapshotSize.Observe(float64(n)) + } +} + func (m *Metrics) recordCacheHit() { if m == nil { return diff --git a/internal/engine/reader_refresh_test.go b/internal/engine/reader_refresh_test.go new file mode 100644 index 0000000..8a8ca0d --- /dev/null +++ b/internal/engine/reader_refresh_test.go @@ -0,0 +1,381 @@ +package engine + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "math" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/cluster" + "github.com/data-accelerator/dart/internal/metrics" + "github.com/data-accelerator/dart/internal/tracker" +) + +type waitingReaderContext struct { + context.Context + once sync.Once + waiting chan struct{} +} + +func (c *waitingReaderContext) Done() <-chan struct{} { + c.once.Do(func() { close(c.waiting) }) + return c.Context.Done() +} + +func refreshClient(t *testing.T, handler http.HandlerFunc) (*Engine, *cluster.View, *metrics.Registry) { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Consume the POST body so net/http can observe a canceled connection + // while a test handler waits. The real tracker also consumes its JOIN. + _, _ = io.Copy(io.Discard, r.Body) + handler(w, r) + })) + t.Cleanup(srv.Close) + p := cluster.NewStaticProvider(cluster.Member{ID: "tracker", Addr: strings.TrimPrefix(srv.URL, "http://"), State: cluster.Ready, Weight: 1}) + reg := metrics.NewRegistry() + e, err := New(Options{Chunk: testCfg(), Store: openStoreAt(t), Fetcher: newFetcher(), SelfID: "reader", Cluster: p, TrackerClient: tracker.NewClient(), Metrics: NewMetrics(reg)}) + if err != nil { + t.Fatal(err) + } + return e, p.Current(), reg +} + +func TestReaderRefreshCoalescesAndCancelsWaiters(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + entered, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + defer unblock() + var calls atomic.Int32 + e, view, reg := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) == 1 { + close(entered) + } + select { + case <-release: + case <-r.Context().Done(): + return + } + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"A", "reader"}, TTLMs: 6000, EpochS: 1}) + }) + done := make(chan []string, 17) + go func() { done <- e.readers(ctx, view, "object", 1) }() + waitBlockEvent(t, ctx, entered) + var waiting []*waitingReaderContext + for i := 0; i < 16; i++ { + wc := &waitingReaderContext{Context: ctx, waiting: make(chan struct{})} + waiting = append(waiting, wc) + go func() { done <- e.readers(wc, view, "object", 1) }() + } + for _, wc := range waiting { + waitBlockEvent(t, ctx, wc.waiting) + } + abandoned, stop := context.WithCancel(ctx) + wc := &waitingReaderContext{Context: abandoned, waiting: make(chan struct{})} + canceled := make(chan []string, 1) + go func() { canceled <- e.readers(wc, view, "object", 1) }() + waitBlockEvent(t, ctx, wc.waiting) + stop() + if got := <-canceled; got != nil { + t.Fatal("canceled waiter returned a snapshot", got) + } + unblock() + for i := 0; i < 17; i++ { + if got := <-done; !reflect.DeepEqual(got, []string{"A", "reader"}) { + t.Error(got) + } + } + if calls.Load() != 1 { + t.Fatal("duplicate JOIN", calls.Load()) + } + var rendered bytes.Buffer + reg.Render(&rendered) + if !strings.Contains(rendered.String(), "dart_reader_snapshot_size_count 1\n") || !strings.Contains(rendered.String(), "dart_reader_snapshot_size_sum 2\n") { + t.Fatal(rendered.String()) + } + // Pin a fresh deadline: this assertion must not depend on the scheduler + // finishing the preceding checks inside the 50 ms startup interval. + e.rsMu.Lock() + fresh := *e.rs["object"] + fresh.expiresAt = time.Now().Add(time.Minute) + e.rs["object"] = &fresh + e.rsMu.Unlock() + e.readers(ctx, view, "object", 1) + if calls.Load() != 1 { + t.Fatal("fresh cache caused JOIN") + } + // Expiration alone does no work; only a subsequent routing lookup refreshes. + e.rsMu.Lock() + old := *e.rs["object"] + old.expiresAt = time.Now().Add(-time.Second) + e.rs["object"] = &old + e.rsMu.Unlock() + time.Sleep(10 * time.Millisecond) + if calls.Load() != 1 { + t.Fatal("background refresh") + } + e.readers(ctx, view, "object", 1) + if calls.Load() != 2 { + t.Fatal("expired view did not refresh") + } +} + +func TestReaderRefreshCanceledLeaderRetries(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + entered := make(chan struct{}) + var calls atomic.Int32 + e, view, _ := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) == 1 { + close(entered) + select { + case <-r.Context().Done(): + case <-ctx.Done(): + } + return + } + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"new"}, TTLMs: 6000}) + }) + leader, stop := context.WithCancel(ctx) + defer stop() + first := make(chan []string, 1) + go func() { first <- e.readers(leader, view, "object", 1) }() + waitBlockEvent(t, ctx, entered) + wc := &waitingReaderContext{Context: ctx, waiting: make(chan struct{})} + second := make(chan []string, 1) + go func() { second <- e.readers(wc, view, "object", 1) }() + waitBlockEvent(t, ctx, wc.waiting) + stop() + if got := <-first; got != nil { + t.Fatal(got) + } + if got := <-second; !reflect.DeepEqual(got, []string{"new"}) { + t.Fatal(got) + } + if calls.Load() != 2 || len(e.rsCalls) != 0 { + t.Fatal("refresh flight leaked or was not replaced", calls.Load()) + } +} + +func TestReaderRefreshOldTrackerCannotOverwriteNew(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + e, oldView, _ := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + close(entered) + select { + case <-release: + case <-r.Context().Done(): + return + } + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"old"}, TTLMs: 6000, EpochS: 999}) + }) + newServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"new"}, TTLMs: 6000, EpochS: 1}) + })) + defer newServer.Close() + newView := cluster.NewStaticProvider(cluster.Member{ID: "new-tracker", Addr: strings.TrimPrefix(newServer.URL, "http://"), State: cluster.Ready, Weight: 1}).Current() + done := make(chan []string, 1) + go func() { done <- e.readers(ctx, oldView, "object", 1) }() + waitBlockEvent(t, ctx, entered) + if got := e.readers(ctx, newView, "object", 1); !reflect.DeepEqual(got, []string{"new"}) { + t.Fatal(got) + } + unblock() + if got := <-done; !reflect.DeepEqual(got, []string{"old"}) { + t.Fatal("old caller lost its view", got) + } + if got := e.rs["object"].nodes; !reflect.DeepEqual(got, []string{"new"}) { + t.Fatal("late old tracker replaced newer view", got) + } +} + +func TestReaderRefreshLeaseBounds(t *testing.T) { + for _, tc := range []struct { + ttl int64 + want time.Duration + }{{0, 4 * time.Second}, {math.MinInt64, 4 * time.Second}, {math.MaxInt64, tracker.MaxLeaseTTL}, {1, time.Millisecond}} { + e, view, _ := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"reader"}, TTLMs: tc.ttl}) + }) + e.readers(context.Background(), view, "object", 1) + rs := e.rs["object"] + if rs == nil || rs.leaseUntil.Sub(rs.cachedAt) != tc.want || rs.interval > tc.want/2 { + t.Fatal(tc, rs) + } + } +} + +func TestReaderRefreshBackoffAndLeaseReset(t *testing.T) { + var calls atomic.Int32 + e, view, _ := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + n := calls.Add(1) + // Change both the published list and its epoch on every refresh. + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"reader", strings.Repeat("A", int(n))}, TTLMs: 6000, EpochS: uint64(n)}) + }) + // Every successful refresh advances solely with time. Changing the reader + // list and publication epoch must not reset this schedule. + for _, want := range []time.Duration{50, 100, 200, 400, 800, 1600, 3000, 3000} { + e.readers(context.Background(), view, "object", 1) + s := e.rs["object"] + if s.interval != want*time.Millisecond { + t.Fatal("interval", s.interval, want) + } + copy := *s + copy.expiresAt = time.Now().Add(-time.Second) + e.rs["object"] = © + } + s := *e.rs["object"] + s.leaseUntil = time.Now().Add(-time.Second) + e.rs["object"] = &s + e.readers(context.Background(), view, "object", 1) + if e.rs["object"].interval != 50*time.Millisecond || calls.Load() != 9 { + t.Fatal("idle lease did not restart schedule") + } + // A short grant wins over the startup interval and the previous backoff. + if d := readerRefreshDelay(0, time.Millisecond); d != time.Millisecond { + t.Fatal(d) + } + if d := readerRefreshDelay(time.Second, time.Millisecond); d != time.Millisecond { + t.Fatal(d) + } +} + +func TestReaderRefreshFailureDoesNotCache(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + var calls atomic.Int32 + e, view, _ := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) == 1 { + close(entered) + select { + case <-release: + case <-r.Context().Done(): + return + } + w.WriteHeader(http.StatusServiceUnavailable) + return + } + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"reader"}, TTLMs: 6000}) + }) + leader := make(chan []string, 1) + go func() { leader <- e.readers(ctx, view, "object", 1) }() + waitBlockEvent(t, ctx, entered) + wc := &waitingReaderContext{Context: ctx, waiting: make(chan struct{})} + waiter := make(chan []string, 1) + go func() { waiter <- e.readers(wc, view, "object", 1) }() + waitBlockEvent(t, ctx, wc.waiting) + unblock() + if a, b := <-leader, <-waiter; a != nil || b != nil { + t.Fatal(a, b) + } + if calls.Load() != 1 || len(e.rsCalls) != 0 || len(e.rs) != 0 { + t.Fatal("failure retained or duplicated") + } + if got := e.readers(ctx, view, "object", 1); !reflect.DeepEqual(got, []string{"reader"}) || calls.Load() != 2 { + t.Fatal("next demand did not retry", got) + } +} + +func cachedReaderEngine(n int) (*Engine, *cluster.View) { + members := make([]cluster.Member, n) + for i := range members { + members[i] = cluster.Member{ID: fmt.Sprintf("node-%04d", i), Addr: fmt.Sprintf("peer-%d:1", i), Weight: 1, State: cluster.Ready} + } + view := cluster.NewView(members) + e := &Engine{selfID: "reader", trackerClient: tracker.NewClient()} + addr, _, _ := e.trackerAddr(view, 1) + now := time.Now() + e.rs = map[string]*readerSet{"object": {nodes: []string{"reader"}, basis: readerBasis{addr: addr, epoch: view.Epoch()}, cachedAt: now, expiresAt: now.Add(time.Hour), leaseUntil: now.Add(time.Hour)}} + e.readers(context.Background(), view, "object", 1) + return e, view +} + +func TestReaderSnapshotCacheHitDoesNotAllocate(t *testing.T) { + e, view := cachedReaderEngine(256) + for _, v := range []*cluster.View{view, cluster.NewView(view.Members())} { + // AllocsPerRun warms once. A replacement view selecting the same + // tracker may update the cache once, then must also avoid ranking. + allocs := testing.AllocsPerRun(100, func() { + if got := e.readers(context.Background(), v, "object", 1); len(got) != 1 { + panic("lost cached snapshot") + } + }) + if allocs != 0 { + t.Fatalf("unchanged-view cache hit allocates: %g", allocs) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if got := e.readers(ctx, v, "object", 1); got != nil { + t.Fatal("canceled lookup returned cached readers") + } + } +} + +func BenchmarkReaderSnapshotCacheHit(b *testing.B) { + for _, n := range []int{32, 256, 4096} { + b.Run(fmt.Sprint(n), func(b *testing.B) { + e, view := cachedReaderEngine(n) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if got := e.readers(context.Background(), view, "object", 1); len(got) != 1 { + b.Fatal("lost cached snapshot") + } + } + }) + } +} + +func TestReaderSnapshotCacheTracksAddressAndLiveness(t *testing.T) { + for _, change := range []string{"address", "liveness"} { + t.Run(change, func(t *testing.T) { + e, oldView, _ := refreshClient(t, func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"old"}, TTLMs: 6000}) + }) + if got := e.readers(context.Background(), oldView, "object", 1); !reflect.DeepEqual(got, []string{"old"}) { + t.Fatal(got) + } + fresh := *e.rs["object"] + fresh.expiresAt = time.Now().Add(time.Minute) + e.rs["object"] = &fresh + members := oldView.Members() + var want []string + if change == "address" { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(tracker.JoinResponse{Readers: []string{"new"}, TTLMs: 6000}) + })) + defer s.Close() + members[0].Addr = s.Listener.Addr().String() + want = []string{"new"} + } else { + members[0].State = cluster.Suspect + } + view := cluster.NewView(members) + if view.Epoch() != oldView.Epoch() { + t.Fatal("test must retain ordering epoch") + } + if got := e.readers(context.Background(), view, "object", 1); !reflect.DeepEqual(got, want) { + t.Fatal("reused stale tracker selection", got, want) + } + }) + } +} diff --git a/internal/engine/readers.go b/internal/engine/readers.go index 84f20a6..f3c5224 100644 --- a/internal/engine/readers.go +++ b/internal/engine/readers.go @@ -9,24 +9,54 @@ import ( "github.com/data-accelerator/dart/internal/tracker" ) -// readerSetTTL is the fallback cache period for a locally cached reader set, -// used only when the tracker granted no lease duration. Normally the cache -// period is half the granted lease (see readers), so renewal always lands -// before the lease lapses. +// readerSetTTL is used when a tracker reports no lease duration. const readerSetTTL = 2 * time.Second -// readerSet is a cached, frozen reader set for one file. +// Startup refreshes are demand-driven: no timer or background worker is kept +// for an object. The normal ceiling remains half of the granted reader lease. +const initialReaderRefresh = 50 * time.Millisecond + +type readerBasis struct { + addr string + epoch uint64 +} + +type readerFlightKey struct { + objectID string + basis readerBasis +} + +type readerFlight struct { + done chan struct{} + leader context.Context + nodes []string +} + +// Published cache entries are immutable; callers only read their node slices. type readerSet struct { - nodes []string - epoch uint64 - cachedAt time.Time - expiresAt time.Time + nodes []string + epoch uint64 + cachedAt, expiresAt time.Time + leaseUntil time.Time + interval time.Duration + basis readerBasis + view *cluster.View // immutable tracker-selection input +} + +func readerRefreshDelay(previous, ceiling time.Duration) time.Duration { + if previous <= 0 { + return min(initialReaderRefresh, ceiling) + } + if previous >= ceiling/2 { + return ceiling + } + return min(previous*2, ceiling) } // trackerAddr returns the address of the tracker responsible for fileKey: the // HRW top-1 Ready member (every node computes the same one, so no registry is -// needed). ok is false when membership is unusable or this node is the tracker -// (in which case the caller uses the local registry directly). +// needed). ok is false when membership is unusable; self selects the local +// registry when this node is the tracker. func (e *Engine) trackerAddr(view *cluster.View, fileKey uint64) (addr string, self bool, ok bool) { ranked := hashring.Rank(fileKey, view.Ready()) if len(ranked) == 0 { @@ -42,71 +72,126 @@ func (e *Engine) trackerAddr(view *cluster.View, fileKey uint64) (addr string, s return m.Addr, false, true } -// readers returns the active reader set for the object, JOINing the tracker (or -// the local registry when this node is the tracker) and caching the result for -// readerSetTTL. It returns nil when no reader set is available, in which case -// callers with tracking enabled route directly to the preferred owner. +// readers returns a cached reader set or joins the tracker. One inline caller +// owns each refresh; waiters can cancel independently, and a canceled leader +// allows a live waiter to retry without letting the old result overwrite it. func (e *Engine) readers(ctx context.Context, view *cluster.View, objectID string, fileKey uint64) []string { if e.trackerClient == nil && e.trackerReg == nil { return nil } + if ctx.Err() != nil { + return nil + } + // Reuse tracker selection only for the identical immutable view. Epoch + // alone excludes address and liveness changes, which can change the tracker. now := time.Now() - e.rsMu.Lock() - if rs, hit := e.rs[objectID]; hit && now.Before(rs.expiresAt) { - nodes := rs.nodes + if cached := e.rs[objectID]; cached != nil && cached.view == view && now.Before(cached.expiresAt) { + nodes := cached.nodes e.rsMu.Unlock() return nodes } e.rsMu.Unlock() - addr, self, ok := e.trackerAddr(view, fileKey) - if !ok { + if !ok || (self && e.trackerReg == nil) || (!self && e.trackerClient == nil) { return nil } - - var resp tracker.JoinResponse - if self { - if e.trackerReg == nil { + basis := readerBasis{addr: addr, epoch: view.Epoch()} + key := readerFlightKey{objectID: objectID, basis: basis} + for { + if ctx.Err() != nil { return nil } - resp = e.trackerReg.Join(objectID, e.selfID, 0) - } else { - if e.trackerClient == nil { - return nil + now := time.Now() + e.rsMu.Lock() + previous := e.rs[objectID] + if previous != nil && previous.basis == basis && now.Before(previous.expiresAt) { + // A newly published view may select the same tracker. Remember it + // without changing the refresh deadline or publication age. + if previous.view != view { + updated := *previous + updated.view = view + e.rs[objectID] = &updated + } + nodes := previous.nodes + e.rsMu.Unlock() + return nodes } - var err error - resp, err = e.trackerClient.Join(ctx, addr, objectID, e.selfID, 0) - if err != nil { - // Tracker unreachable: route directly to the owner rather than - // failing the read (S is soft state). - return nil + if c := e.rsCalls[key]; c != nil && c.leader.Err() == nil { + e.rsMu.Unlock() + select { + case <-ctx.Done(): + return nil + case <-c.leader.Done(): + continue + case <-c.done: + if ctx.Err() != nil { + return nil + } + if c.leader.Err() != nil { + continue + } + return c.nodes + } } - } + if e.rsCalls == nil { + e.rsCalls = make(map[readerFlightKey]*readerFlight) + } + c := &readerFlight{done: make(chan struct{}), leader: ctx} + e.rsCalls[key] = c + e.rsMu.Unlock() - // Cache for half the granted lease, so the next re-Join lands well before - // our tracker lease lapses: a renewal period longer than the lease would - // silently drop this node from the frozen reader set while it is still - // reading. (readerSetTTL is only the fallback for a tracker that reported - // no grant.) - ttl := time.Duration(resp.TTLMs) * time.Millisecond / 2 - if ttl <= 0 { - ttl = readerSetTTL - } - e.rsMu.Lock() - if len(e.rs) > 256 { - // Bound the cache: drop expired entries before growing past a few - // hundred objects. Live entries stay; the map tracks objects currently - // being read, which is naturally small. - for id, rs := range e.rs { - if now.After(rs.expiresAt) { - delete(e.rs, id) + var resp tracker.JoinResponse + var err error + if self { + resp = e.trackerReg.Join(objectID, e.selfID, 0) + } else { + resp, err = e.trackerClient.Join(ctx, addr, objectID, e.selfID, 0) + } + e.rsMu.Lock() + if e.rsCalls[key] == c { + delete(e.rsCalls, key) + if err == nil && ctx.Err() == nil { + // Clamp before duration conversion, matching the registry's + // grant bound. A missing grant retains the historical fallback. + lease := time.Duration(0) + if resp.TTLMs > 0 { + lease = time.Duration(min(resp.TTLMs, tracker.MaxLeaseTTL.Milliseconds())) * time.Millisecond + } + ceiling := lease / 2 + if ceiling <= 0 { + ceiling = readerSetTTL + lease = 2 * ceiling + } + interval := time.Duration(0) + if previous != nil && previous.basis == basis && now.Before(previous.leaseUntil) { + interval = previous.interval + } + interval = readerRefreshDelay(interval, ceiling) + if len(e.rs) > 256 { + for id, cached := range e.rs { + if !now.Before(cached.leaseUntil) { + delete(e.rs, id) + } + } + } + if e.rs == nil { + e.rs = make(map[string]*readerSet) + } + // A response from an older tracker view must not replace a cache + // installed by a more recently started refresh for this object. + if current := e.rs[objectID]; current == nil || !current.cachedAt.After(now) { + e.rs[objectID] = &readerSet{nodes: resp.Readers, epoch: resp.EpochS, cachedAt: now, + expiresAt: now.Add(interval), leaseUntil: now.Add(lease), interval: interval, basis: basis, view: view} + } + e.mx.recordReaderSnapshot(len(resp.Readers)) + c.nodes = resp.Readers } } + close(c.done) + e.rsMu.Unlock() + return c.nodes } - e.rs[objectID] = &readerSet{nodes: resp.Readers, epoch: resp.EpochS, cachedAt: now, expiresAt: now.Add(ttl)} - e.rsMu.Unlock() - return resp.Readers } // treeNodes returns the ordered candidate set the distribution tree is built diff --git a/internal/engine/readers_test.go b/internal/engine/readers_test.go index 367dc59..3527f26 100644 --- a/internal/engine/readers_test.go +++ b/internal/engine/readers_test.go @@ -63,11 +63,9 @@ func TestTreeNodesPrefersReaderSet(t *testing.T) { t.Fatalf("fileKey mismatch: %d vs %d", got, fileKey) } - // Register the two readers and let the tick publish them. + // Both readers are published in their JOIN responses. reg.Join(oid, self, 0) reg.Join(oid, other, 0) - time.Sleep(5 * time.Millisecond) - reg.Join(oid, other, 0) nodes, fromReaders := e.treeNodes(context.Background(), view, oid, fileKey, 12345) if !fromReaders { @@ -230,7 +228,11 @@ func TestReaderSnapshotChurnPreservesStrictOrder(t *testing.T) { } } before := append([]string(nil), ids...) - e.rs["object"] = &readerSet{nodes: ids, expiresAt: time.Now().Add(time.Hour)} + addr, _, _ := e.trackerAddr(view, 1) + e.rs["object"] = &readerSet{nodes: ids, expiresAt: time.Now().Add(time.Hour), basis: readerBasis{addr: addr, epoch: view.Epoch()}, view: view} + if got := e.readers(context.Background(), view, "object", 1); !reflect.DeepEqual(got, ids) { + t.Fatal("fixture snapshot was bypassed", got, ids) + } order, _ := e.treeNodes(context.Background(), view, "object", 1, key) seen, selfIndex := make(map[string]bool), -1 for i, n := range order { diff --git a/internal/engine/review_regression_test.go b/internal/engine/review_regression_test.go index aa9acd6..d46b5ef 100644 --- a/internal/engine/review_regression_test.go +++ b/internal/engine/review_regression_test.go @@ -156,6 +156,9 @@ func TestRootFallbackAfterHedgeAdmissionDeclines(t *testing.T) { if rootCalls.Load() != 0 { t.Fatal("root contacted without admission") } + if mx.peerFetchRoot.Value() != 0 || mx.peerFetchParent.Value() != 1 { + t.Fatal("unadmitted hedge counted as dispatched", mx.peerFetchRoot.Value(), mx.peerFetchParent.Value()) + } unblock() if err := <-done; err != nil { t.Fatal("healthy root was skipped after capacity returned", err) @@ -163,6 +166,9 @@ func TestRootFallbackAfterHedgeAdmissionDeclines(t *testing.T) { if rootCalls.Load() != 1 || parentCalls.Load() != 1 { t.Fatal("wrong attempts", rootCalls.Load(), parentCalls.Load()) } + if mx.peerFetchRoot.Value() != 1 || mx.peerFetchParent.Value() != 1 || mx.peerFetchAncestor.Value() != 0 { + t.Fatal("fallback target metrics disagree with HTTP calls") + } if n, b := receiveUsage(e); n != 0 || b != 0 { t.Fatal("receive leak", n, b) } diff --git a/internal/engine/roles_golden_test.go b/internal/engine/roles_golden_test.go index e9ba452..793e858 100644 --- a/internal/engine/roles_golden_test.go +++ b/internal/engine/roles_golden_test.go @@ -76,7 +76,13 @@ func TestRolePlacementAndTreeGolden(t *testing.T) { } readers := append([]string(nil), tc.Readers...) rng.Shuffle(len(readers), func(i, j int) { readers[i], readers[j] = readers[j], readers[i] }) - e.rs["object"] = &readerSet{nodes: readers, expiresAt: time.Now().Add(time.Hour)} + addr, _, _ := e.trackerAddr(view, 123) + e.rs["object"] = &readerSet{nodes: readers, expiresAt: time.Now().Add(time.Hour), basis: readerBasis{addr: addr, epoch: view.Epoch()}, view: view} + if tc.Readers != nil { + if got := e.readers(context.Background(), view, "object", 123); !reflect.DeepEqual(got, readers) { + t.Fatal("golden snapshot was bypassed", got, readers) + } + } if roots := roleIDs(hashring.Rank(tc.Key, view.Roots())); !reflect.DeepEqual(roots, tc.Roots) { t.Fatalf("key=%x roots=%v want=%v", tc.Key, roots, tc.Roots) } diff --git a/internal/engine/routing.go b/internal/engine/routing.go index b10f714..5ebd76d 100644 --- a/internal/engine/routing.go +++ b/internal/engine/routing.go @@ -54,6 +54,22 @@ func (e *Engine) loadRouted(ctx context.Context, url, objectID string, size, ind } ranked, _ := e.treeNodes(ctx, view, objectID, e.fileKey(objectID), key.Chunk) primary, backup := e.hedgeTargets(ranked, view) + // Classify against the direct tree parent, not the first healthy ancestor: + // a breaker-promoted target still counts as ancestor traffic. + parentAddr := "" + if e.mx != nil { + for i, n := range ranked { + if n.ID != e.selfID { + continue + } + if p := hashring.ParentWithRootFanout(i, len(ranked), e.fanout, e.rootFanout); p >= 0 { + if m, ok := view.Get(ranked[p].ID); ok { + parentAddr = m.Addr + } + } + break + } + } attempted := make(map[string]bool) var attemptsMu sync.Mutex rootAddr := "" @@ -75,7 +91,11 @@ func (e *Engine) loadRouted(ctx context.Context, url, objectID string, size, ind if already { return peerResult{} } - data, held, err := e.peer.GetInto(ctx, addr, peer.BlockRequest{Key: key, URL: url, Hop: hop + 1, Supply: supply, Epoch: view.Epoch(), HasEpoch: true}, b.bytes(), nil) + var onStart func() + if e.mx != nil { + onStart = func() { e.mx.recordPeerFetch(supply, addr == parentAddr) } + } + data, held, err := e.peer.GetIntoObserved(ctx, addr, peer.BlockRequest{Key: key, URL: url, Hop: hop + 1, Supply: supply, Epoch: view.Epoch(), HasEpoch: true}, b.bytes(), nil, onStart) if err != nil || !held { return peerResult{held: held, err: err} } diff --git a/internal/engine/routing_metrics_test.go b/internal/engine/routing_metrics_test.go new file mode 100644 index 0000000..42e2b48 --- /dev/null +++ b/internal/engine/routing_metrics_test.go @@ -0,0 +1,72 @@ +package engine + +import ( + "bytes" + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/data-accelerator/dart/internal/chunk" + "github.com/data-accelerator/dart/internal/cluster" + "github.com/data-accelerator/dart/internal/hashring" + "github.com/data-accelerator/dart/internal/metrics" + "github.com/data-accelerator/dart/internal/peer" +) + +func TestPeerFetchTargetMetrics(t *testing.T) { + // The first three nodes form a chain above self. A failed parent and an + // unavailable ancestor lead to a terminal root, with or without the parent + // initially excluded by the breaker. + for _, skipParent := range []bool{false, true} { + t.Run(fmt.Sprint(skipParent), func(t *testing.T) { + const url = "http://origin.invalid/target-metrics" + key := blockKeyFor(url) + oid, _ := chunk.ObjectID(url) + order := hashring.Rank(key.Chunk, []hashring.Node{{ID: "A", Weight: 1}, {ID: "B", Weight: 1}, {ID: "C", Weight: 1}}) + var parentCalls, ancestorCalls, rootCalls atomic.Int32 + parent := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { parentCalls.Add(1); w.WriteHeader(404) })) + defer parent.Close() + ancestor := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ancestorCalls.Add(1); w.WriteHeader(404) })) + defer ancestor.Close() + content := blob(16) + root := httptest.NewServer(&peer.Server{Src: func(_ context.Context, req peer.BlockRequest) ([]byte, bool, error) { + rootCalls.Add(1) + if !req.Supply { + t.Error("root attempt was not terminal") + } + return content, true, nil + }}) + defer root.Close() + p := cluster.NewStaticProvider( + cluster.Member{ID: "R", Addr: root.Listener.Addr().String(), Weight: 1, State: cluster.Ready, Role: cluster.RootOnly}, + cluster.Member{ID: order[0].ID, Addr: ancestor.Listener.Addr().String(), Weight: 1, State: cluster.Ready, Role: cluster.NonRoot}, + cluster.Member{ID: order[1].ID, Addr: parent.Listener.Addr().String(), Weight: 1, State: cluster.Ready, Role: cluster.NonRoot}, + cluster.Member{ID: order[2].ID, Addr: "unused:1", Weight: 1, State: cluster.Ready, Role: cluster.NonRoot}) + pc := peer.NewClient() + pc.Breaker = peer.NewBreaker(peer.BreakerOptions{}) + if skipParent { + pc.Breaker.RecordHardFailure(parent.Listener.Addr().String()) + } + mx := NewMetrics(metrics.NewRegistry()) + e, err := New(Options{Chunk: testCfg(), Store: openStoreAt(t), Fetcher: &forbiddenOrigin{}, Cluster: p, Peer: pc, SelfID: order[2].ID, Role: cluster.NonRoot, Fanout: 1, Metrics: mx}) + if err != nil { + t.Fatal(err) + } + l, err := e.borrowBlock(context.Background(), url, oid, 16, 0, 0) + defer l.Release() + if err != nil || !bytes.Equal(l.Bytes(), content) { + t.Fatal(err) + } + wantParent := uint64(1) + if skipParent { + wantParent = 0 + } + if mx.peerFetchRoot.Value() != 1 || mx.peerFetchParent.Value() != wantParent || mx.peerFetchAncestor.Value() != 1 || rootCalls.Load() != 1 || uint64(parentCalls.Load()) != wantParent || ancestorCalls.Load() != 1 { + t.Fatal("target counts", mx.peerFetchRoot.Value(), mx.peerFetchParent.Value(), mx.peerFetchAncestor.Value(), rootCalls.Load(), parentCalls.Load(), ancestorCalls.Load()) + } + }) + } +} diff --git a/internal/engine/tracker_experiment_test.go b/internal/engine/tracker_experiment_test.go new file mode 100644 index 0000000..027964e --- /dev/null +++ b/internal/engine/tracker_experiment_test.go @@ -0,0 +1,90 @@ +package engine + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "sort" + "sync" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/chunk" +) + +// Opt-in traffic experiment, not a timing assertion. A serialized 40 ms root +// service per 4 MiB block models a finite root bottleneck. Registry time is held +// within the initial tick to isolate publication from expiry. Real HTTP JOINs, +// cache deadlines and payload transfers remain active. No origin is contacted. +func TestTrackerStartupExperiment(t *testing.T) { + if os.Getenv("DART_TRACKER_STARTUP_EXPERIMENT") != "1" { + t.Skip("opt-in startup traffic experiment") + } + for _, scenario := range []struct { + name string + order []int + gap time.Duration + }{ + {"burst", []int{7, 0, 4, 2, 6, 1, 5, 3}, time.Millisecond}, + {"staggered", []int{7, 0, 4, 2, 6, 1, 5, 3}, 30 * time.Millisecond}, + {"reverse", []int{7, 6, 5, 4, 3, 2, 1, 0}, 30 * time.Millisecond}, + {"slow-growth", []int{7, 6, 5, 4, 3, 2, 1, 0}, 120 * time.Millisecond}, + } { + t.Run(scenario.name, func(t *testing.T) { + s := newStartupCluster(t, 8, 40*time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + var wg sync.WaitGroup + var mu sync.Mutex + var latencies []float64 + start := time.Now() + for slot, index := range scenario.order { + wg.Add(1) + go func(slot, index int) { + defer wg.Done() + timer := time.NewTimer(time.Until(start.Add(time.Duration(slot) * scenario.gap))) + defer timer.Stop() + select { + case <-timer.C: + case <-ctx.Done(): + return + } + for off := int64(0); off < int64(len(s.content)); off += chunk.MiB { + var out bytes.Buffer + began := time.Now() + err := s.readers[index].Serve(ctx, &out, s.url, off, off+chunk.MiB-1) + elapsed := time.Since(began).Seconds() * 1000 + if err != nil || !bytes.Equal(out.Bytes(), s.content[off:off+chunk.MiB]) { + t.Errorf("reader %d offset %d: %v", index, off, err) + return + } + mu.Lock() + latencies = append(latencies, elapsed) + mu.Unlock() + } + }(slot, index) + } + wg.Wait() + elapsed := time.Since(start).Seconds() + s.closeHTTP() + if elapsed+0.001 < float64(s.rootBytes.Load())/float64(4*chunk.MiB)*0.040 { + t.Fatal("serialized root service limit was not exercised") + } + if len(latencies) != 8*32 { + t.Fatalf("incomplete reads: %d", len(latencies)) + } + sort.Float64s(latencies) + result := map[string]any{ + "scenario": scenario.name, "policy": os.Getenv("DART_REFRESH_POLICY"), + "seconds": elapsed, "aggregate_gbps": float64(8*len(s.content)) * 8 / elapsed / 1e9, + "root_mib": float64(s.rootBytes.Load()) / float64(chunk.MiB), "peer_mib": float64(s.peerBytes.Load()) / float64(chunk.MiB), + "joins": s.joins.Load(), "join_response_bytes": s.joinBytes.Load(), "join_handler_ms": float64(s.joinNanos.Load()) / 1e6, + "range_p50_ms": latencies[len(latencies)/2], "range_p95_ms": latencies[len(latencies)*95/100], "range_p99_ms": latencies[len(latencies)*99/100], + } + encoded, _ := json.Marshal(result) + fmt.Printf("STARTUP %s\n", encoded) + }) + } +} diff --git a/internal/engine/tracker_startup_test.go b/internal/engine/tracker_startup_test.go new file mode 100644 index 0000000..1e54725 --- /dev/null +++ b/internal/engine/tracker_startup_test.go @@ -0,0 +1,150 @@ +package engine + +import ( + "bytes" + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/chunk" + "github.com/data-accelerator/dart/internal/cluster" + "github.com/data-accelerator/dart/internal/fetch" + "github.com/data-accelerator/dart/internal/hashring" + "github.com/data-accelerator/dart/internal/peer" + "github.com/data-accelerator/dart/internal/store" + "github.com/data-accelerator/dart/internal/tracker" +) + +type startupCluster struct { + url, oid string + content []byte + readers []*Engine // HRW order, excluding the root + servers []*httptest.Server + rootBytes, peerBytes, joins, joinBytes atomic.Int64 + joinNanos atomic.Int64 + registry *tracker.Registry + trackerTime time.Time +} + +// The tracker clock is fixed: the initial tick can never hide a JOIN publication +// bug. Every reader uses the real HTTP tracker client and its real snapshot cache. +func newStartupCluster(t *testing.T, blocks int, rootDelay time.Duration) *startupCluster { + t.Helper() + const block = 4 * chunk.MiB + s := &startupCluster{url: "http://origin.invalid/startup", content: blob(int(block) * blocks), trackerTime: time.Now()} + s.oid, _ = chunk.ObjectID(s.url) + cfg := chunk.Config{ChunkSize: 256 * chunk.MiB, BlockSize: block} + ck := chunk.ChunkKey("dart", s.oid, 0) + fk := chunk.ChunkKey("dart", s.oid, -1) + nodes := make([]hashring.Node, 8) + for i := range nodes { + nodes[i] = hashring.Node{ID: fmt.Sprint("reader-", i), Weight: 1} + } + rootID := "" + for i := 0; ; i++ { + id := fmt.Sprint("root-", i) + if hashring.Rank(fk, append(append([]hashring.Node(nil), nodes...), hashring.Node{ID: id, Weight: 1}))[0].ID == id { + rootID = id + break + } + } + s.registry = tracker.NewRegistry(tracker.Options{Now: func() time.Time { return s.trackerTime }}) + open := func() *store.MemStore { + st, err := store.OpenMem(store.MemOptions{SlotSize: block, Slots: blocks}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st + } + root := open() + for i := 0; i < blocks; i++ { + if err := root.Put(store.BlockKey{Chunk: ck, Block: uint64(i)}, s.content[int(block)*i:int(block)*(i+1)]); err != nil { + t.Fatal(err) + } + } + mux := http.NewServeMux() + th := (&tracker.Server{R: s.registry}).Handler() + mux.Handle("/tracker/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.joins.Add(1) + start := time.Now() + defer func() { s.joinNanos.Add(time.Since(start).Nanoseconds()) }() + th.ServeHTTP(countedPeerWriter{w, &s.joinBytes}, r) + })) + ph := &peer.Server{Borrow: func(_ context.Context, req peer.BlockRequest) (store.Lease, bool, error) { + return root.Acquire(req.Key) + }} + rootSlot := make(chan struct{}, 1) + mux.Handle("/peer/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if rootDelay > 0 { + select { + case rootSlot <- struct{}{}: + case <-r.Context().Done(): + return + } + defer func() { <-rootSlot }() + timer := time.NewTimer(rootDelay) + defer timer.Stop() + select { + case <-timer.C: + case <-r.Context().Done(): + return + } + } + ph.ServeHTTP(countedPeerWriter{w, &s.rootBytes}, r) + })) + rs := httptest.NewServer(mux) + t.Cleanup(rs.Close) + s.servers = append(s.servers, rs) + members := []cluster.Member{{ID: rootID, Addr: strings.TrimPrefix(rs.URL, "http://"), Weight: 1, State: cluster.Ready, Role: cluster.RootOnly}} + p := cluster.NewStaticProvider() + for _, node := range hashring.Rank(ck, nodes) { + e, err := New(Options{Chunk: cfg, Store: open(), Fetcher: &forbiddenOrigin{}, Cluster: p, Peer: peer.NewClient(), SelfID: node.ID, Role: cluster.NonRoot, TrackerClient: tracker.NewClient(), Hedge: false, PrefetchWindow: 1}) + if err != nil { + t.Fatal(err) + } + e.sizes[fetch.OriginKey(s.url)] = sizeMeta{size: int64(len(s.content))} + s.readers = append(s.readers, e) + h := &peer.Server{Borrow: e.PeerLeaseSource()} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { h.ServeHTTP(countedPeerWriter{w, &s.peerBytes}, r) })) + t.Cleanup(srv.Close) + s.servers = append(s.servers, srv) + members = append(members, cluster.Member{ID: node.ID, Addr: strings.TrimPrefix(srv.URL, "http://"), Weight: 1, State: cluster.Ready, Role: cluster.NonRoot}) + } + p.Set(members) + return s +} + +func (s *startupCluster) closeHTTP() { + for _, srv := range s.servers { + srv.Close() + } +} + +func TestTrackerStartupPublishesBeforeTick(t *testing.T) { + s := newStartupCluster(t, 1, 0) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + // The worst-ranked first reader causes every later reader to choose the root + // when all JOIN responses remain frozen at that singleton. Immediate growth + // instead lets ranks 1..6 discover rank 0, without refreshing their predecessors. + for _, i := range []int{7, 0, 1, 2, 3, 4, 5, 6} { + var out bytes.Buffer + if err := s.readers[i].Serve(ctx, &out, s.url, 0, int64(len(s.content)-1)); err != nil { + t.Fatal(err) + } + if !bytes.Equal(out.Bytes(), s.content) { + t.Fatal("wrong bytes") + } + } + s.closeHTTP() // Join sender handlers before inspecting completed Write counts. + t.Logf("root_bytes=%d peer_bytes=%d joins=%d", s.rootBytes.Load(), s.peerBytes.Load(), s.joins.Load()) + if s.rootBytes.Load() != 2*4*chunk.MiB || s.peerBytes.Load() != 6*4*chunk.MiB || s.joins.Load() != 8 { + t.Fatal("startup did not use the growing reader set") + } +} diff --git a/internal/node/node.go b/internal/node/node.go index f4a7bd9..3f2b7bf 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -462,7 +462,7 @@ func parseFlags(args []string, out io.Writer, schemes []DiscoveryScheme) (config fs.IntVar(&cfg.rootFanout, "root-fanout", 1, "preferred root branching factor (normal tree children, not a transfer limit)") fs.StringVar(&cfg.adminAddr, "admin", ":9100", "admin/metrics listen address; empty disables") fs.BoolVar(&cfg.readerTree, "reader-tree", true, "build the distribution tree over the active reader set (per-file tracker)") - fs.DurationVar(&cfg.trackerTick, "tracker-tick", tracker.DefaultTick, "how often a tracker republishes a reader set") + fs.DurationVar(&cfg.trackerTick, "tracker-tick", tracker.DefaultTick, "tracker removal/expiry batching interval (new readers publish immediately)") fs.IntVar(&cfg.replicas, "replicas", 1, "HRW candidates that authoritatively hold a chunk (the owned budget)") fs.Float64Var(&cfg.ownedFraction, "owned-fraction", 0.8, "owned cache share in [0,1]; 0 borrowed-only, 1 owned-only") sizeVar(fs, &cfg.memSize, "mem-size", 256*chunk.MiB, "in-memory hot-set size; below one block disables the memory tier") diff --git a/internal/peer/observed_test.go b/internal/peer/observed_test.go new file mode 100644 index 0000000..9ad270c --- /dev/null +++ b/internal/peer/observed_test.go @@ -0,0 +1,93 @@ +package peer + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func TestGetIntoObservedDispatch(t *testing.T) { + for _, status := range []int{200, 404, 503} { + var notified, served atomic.Int32 + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + served.Add(1) + if notified.Load() != 1 { + t.Error("dispatch notification did not precede HTTP request") + } + w.WriteHeader(status) + if status == 200 { + _, _ = w.Write([]byte("data")) + } + })) + c := NewClient() + _, held, err := c.GetIntoObserved(context.Background(), s.Listener.Addr().String(), BlockRequest{}, make([]byte, 4), nil, func() { notified.Add(1) }) + s.Close() + if notified.Load() != 1 || served.Load() != 1 || held != (status == 200) || (err != nil) != (status == 503) { + t.Fatal(status, notified.Load(), served.Load(), held, err) + } + } + for _, mode := range []string{"breaker", "invalid", "canceled", "transport"} { + t.Run(mode, func(t *testing.T) { + c := NewClient() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + addr := "peer:1" + var calls, notified int + c.HTTP = &http.Client{Transport: boundRoundTripFunc(func(*http.Request) (*http.Response, error) { + calls++ + if notified != 1 { + t.Error("transport started before notification") + } + return nil, errors.New("transport unavailable") + })} + switch mode { + case "breaker": + c.Breaker = NewBreaker(BreakerOptions{}) + c.Breaker.RecordHardFailure(addr) + case "invalid": + addr = "bad\naddress" + case "canceled": + cancel() + } + _, _, err := c.GetIntoObserved(ctx, addr, BlockRequest{}, make([]byte, 4), nil, func() { notified++ }) + want := 0 + if mode == "transport" { + want = 1 + } + if err == nil || calls != want || notified != want { + t.Fatal(err, calls, notified) + } + }) + } +} + +func TestGetIntoObservedClientTimeoutBeforeDispatch(t *testing.T) { + var requests, notified atomic.Int32 + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + w.WriteHeader(http.StatusNotFound) + })) + defer s.Close() + now := time.Unix(1000, 0) + b := NewBreaker(BreakerOptions{Cooldown: time.Second, Now: func() time.Time { return now }}) + addr := s.Listener.Addr().String() + b.RecordHardFailure(addr) + now = now.Add(time.Second) + c := NewClient() + c.Breaker = b + // The client-owned deadline has elapsed before HTTP dispatch, even though + // the original caller remains live. This must not prove peer recovery. + c.Timeout = time.Nanosecond + caller := context.Background() + _, _, err := c.GetIntoObserved(caller, addr, BlockRequest{}, make([]byte, 4), nil, func() { notified.Add(1) }) + if !errors.Is(err, context.DeadlineExceeded) || caller.Err() != nil { + t.Fatal("wrong deadline source", err, caller.Err()) + } + if requests.Load() != 0 || notified.Load() != 0 || b.State(addr) != BreakerOpen { + t.Fatal("undispatched client timeout claimed recovery", requests.Load(), notified.Load(), b.State(addr)) + } +} diff --git a/internal/peer/peer.go b/internal/peer/peer.go index 75396c5..ce98efb 100644 --- a/internal/peer/peer.go +++ b/internal/peer/peer.go @@ -413,7 +413,7 @@ func (c *Client) record(addr string, permit breakerPermit, o outcome) { // Get fetches a block from the peer at addr (host:port) per req. held is false // if the peer returned 404 (cannot provide the block). func (c *Client) Get(ctx context.Context, addr string, req BlockRequest) (data []byte, held bool, err error) { - return c.get(ctx, addr, req, -1, nil, nil) + return c.get(ctx, addr, req, -1, nil, nil, nil) } // GetValidated rejects negative expected lengths and reads at most expected+1 @@ -423,13 +423,13 @@ func (c *Client) GetValidated(ctx context.Context, addr string, req BlockRequest if expected < 0 { return nil, false, errors.New("peer: negative expected block length") } - return c.get(ctx, addr, req, expected, validate, nil) + return c.get(ctx, addr, req, expected, validate, nil, nil) } // ErrBlockLength reports a peer body that does not match the expected geometry. var ErrBlockLength = errors.New("peer: invalid block length") -func (c *Client) get(ctx context.Context, addr string, req BlockRequest, expected int64, validate func(context.Context, int64) error, dst []byte) (data []byte, held bool, err error) { +func (c *Client) get(ctx context.Context, addr string, req BlockRequest, expected int64, validate func(context.Context, int64) error, dst []byte, onStart func()) (data []byte, held bool, err error) { permit, err := c.allow(addr) if err != nil { return nil, false, err @@ -457,6 +457,19 @@ func (c *Client) get(ctx context.Context, addr string, req BlockRequest, expecte if client == nil { client = http.DefaultClient } + if err := ctx.Err(); err != nil { + // Preserve the same caller/client-timeout distinction as Do failures. + // A client-owned deadline is not evidence that this peer answered. + if caller.Err() != nil { + c.record(addr, permit, outcomeAnswered) + } else { + c.record(addr, permit, classify(err)) + } + return nil, false, err + } + if onStart != nil { + onStart() + } resp, err := client.Do(hreq) if err != nil { // A caller-aborted request (hedging loser, client disconnect) is not a @@ -527,5 +540,13 @@ func (c *Client) get(ctx context.Context, addr string, req BlockRequest, expecte // dst may contain partial bytes and must not be published. The caller owns dst; // it is never retained after this call, including on cancellation. func (c *Client) GetInto(ctx context.Context, addr string, req BlockRequest, dst []byte, validate func(context.Context, int64) error) ([]byte, bool, error) { - return c.get(ctx, addr, req, int64(len(dst)), validate, dst) + return c.get(ctx, addr, req, int64(len(dst)), validate, dst, nil) +} + +// GetIntoObserved is GetInto with an inline, nonblocking notification immediately +// before dispatching an HTTP attempt. Breaker rejection, invalid request setup, +// and already-canceled contexts do not notify. Dispatch may still fail before +// bytes reach the network; internal HTTP transport retries are not new attempts. +func (c *Client) GetIntoObserved(ctx context.Context, addr string, req BlockRequest, dst []byte, validate func(context.Context, int64) error, onStart func()) ([]byte, bool, error) { + return c.get(ctx, addr, req, int64(len(dst)), validate, dst, onStart) } diff --git a/internal/tracker/client.go b/internal/tracker/client.go index 1d34830..b569dfa 100644 --- a/internal/tracker/client.go +++ b/internal/tracker/client.go @@ -29,7 +29,7 @@ func (c *Client) httpClient() *http.Client { return http.DefaultClient } -// Join announces interest in file and returns the frozen reader set. +// Join announces interest in file and returns the published reader set. func (c *Client) Join(ctx context.Context, addr, file, node string, ttl time.Duration) (JoinResponse, error) { body, err := json.Marshal(JoinRequest{File: file, Node: node, TTLMs: ttl.Milliseconds()}) if err != nil { diff --git a/internal/tracker/tracker.go b/internal/tracker/tracker.go index 23fd8a7..aa67d36 100644 --- a/internal/tracker/tracker.go +++ b/internal/tracker/tracker.go @@ -1,18 +1,17 @@ // Package tracker maintains the per-file active reader set S: which nodes are // currently reading a given object. // -// The distribution tree is built over S rather than over all members, so a -// node's parent is always another reader (and therefore a node that actually -// wants the data) instead of an arbitrary member that would have to -// fetch-on-behalf. See docs/tracker.md. +// Below the fixed chunk owner, the distribution tree is built over S rather +// than all members. Intermediate nodes are readers that want the data, instead +// of arbitrary members fetching only on behalf of others. See docs/tracker.md. // // Design properties: // // - Leases: a reader JOINs with a TTL and refreshes; when it stops reading the // lease lapses and it drops out of S, shrinking the tree. S is soft state, so // a tracker restart self-heals as readers renew. -// - Tick freeze: the set is only recomputed on a fixed tick, so the topology -// (and epochS) is stable between ticks and TCP connections do not churn. +// - Immediate growth: each new reader is published during its JOIN. Removal +// and expiry are batched on a tick; renewing a lease does not change epochS. // - Control plane only: small JSON messages, no data flows through a tracker. // // Which node is the tracker for a file is decided by the caller (HRW top-1 over @@ -52,12 +51,12 @@ type JoinRequest struct { TTLMs int64 `json:"ttlMs,omitempty"` } -// JoinResponse is the frozen reader set for a file. +// JoinResponse is the published reader set for a file. type JoinResponse struct { - // EpochS changes whenever the frozen set changes; readers compare it to + // EpochS changes whenever the published set changes; readers compare it to // detect a topology change without diffing the member list. EpochS uint64 `json:"epochS"` - // Readers is the frozen set, sorted by node ID (deterministic across nodes). + // Readers is the published set, sorted by node ID (deterministic across nodes). Readers []string `json:"readers"` // TTLMs is the granted lease in milliseconds. TTLMs int64 `json:"ttlMs"` @@ -69,10 +68,10 @@ type lease struct { expireAt time.Time } -// fileState is the live and frozen reader sets for one file. +// fileState is the live and published reader sets for one file. type fileState struct { leases map[string]*lease // node -> lease (live, updated on every join) - frozen []string // set published to readers, recomputed on a tick + frozen []string // sorted published set; grows on JOIN, shrinks on a tick epoch uint64 // bumped when frozen changes nextAt time.Time // when the next recompute is due // lastActivity is the last Join/Readers for this file; it drives idle @@ -96,7 +95,7 @@ type Registry struct { // Options configures a Registry. Zero values select the defaults. type Options struct { - // Tick is how often the frozen set is recomputed. + // Tick is the removal/expiry batching interval; new readers publish immediately. Tick time.Duration // LeaseTTL is the default lease granted to a reader. LeaseTTL time.Duration @@ -133,7 +132,7 @@ func NewRegistry(opt Options) *Registry { } // Join records (or refreshes) node's interest in file and returns the currently -// frozen reader set. The returned readers slice is a copy owned by the caller. +// published reader set. The returned readers slice is a copy owned by the caller. func (r *Registry) Join(file, node string, ttl time.Duration) JoinResponse { if ttl <= 0 { ttl = r.ttl @@ -158,6 +157,16 @@ func (r *Registry) Join(file, node string, ttl time.Duration) JoinResponse { fs.leases[node] = &lease{node: node, expireAt: now.Add(ttl)} } r.refreshLocked(fs, now) + // Growth must be visible in this JOIN response. Only departures need tick + // batching: stale subsets still have strictly ordered forwarding edges. + // Insert into the sorted snapshot without rebuilding it on every renewal. + i := sort.SearchStrings(fs.frozen, node) + if i == len(fs.frozen) || fs.frozen[i] != node { + fs.frozen = append(fs.frozen, "") + copy(fs.frozen[i+1:], fs.frozen[i:]) + fs.frozen[i] = node + fs.epoch++ + } return JoinResponse{ EpochS: fs.epoch, @@ -167,7 +176,7 @@ func (r *Registry) Join(file, node string, ttl time.Duration) JoinResponse { } // Leave drops node's lease on file immediately (a clean shutdown or a reader -// that finished). The frozen set updates on the next tick. +// that finished). The published set updates on the next tick. func (r *Registry) Leave(file, node string) { r.mu.Lock() defer r.mu.Unlock() @@ -179,7 +188,7 @@ func (r *Registry) Leave(file, node string) { } } -// Readers returns the frozen reader set for file (a copy) and its epoch. +// Readers returns the published reader set for file (a copy) and its epoch. func (r *Registry) Readers(file string) ([]string, uint64) { now := r.now() r.mu.Lock() @@ -226,7 +235,7 @@ func (r *Registry) sweepLocked(now time.Time) { } } -// refreshLocked recomputes the frozen set if the tick is due, dropping expired +// refreshLocked recomputes the published set if the tick is due, dropping expired // leases. Caller must hold r.mu. func (r *Registry) refreshLocked(fs *fileState, now time.Time) { if fs.nextAt.IsZero() { @@ -237,13 +246,13 @@ func (r *Registry) refreshLocked(fs *fileState, now time.Time) { return } if now.Before(fs.nextAt) { - return // frozen: keep the topology stable between ticks + return // Departures remain batched; JOIN publishes growth separately. } fs.nextAt = now.Add(r.tick) fs.recompute(now) } -// recompute drops expired leases and rebuilds the frozen set, bumping the epoch +// recompute drops expired leases and rebuilds the published set, bumping the epoch // when it changed. func (fs *fileState) recompute(now time.Time) { live := make([]string, 0, len(fs.leases)) diff --git a/internal/tracker/tracker_test.go b/internal/tracker/tracker_test.go index 966c134..c755b65 100644 --- a/internal/tracker/tracker_test.go +++ b/internal/tracker/tracker_test.go @@ -52,30 +52,44 @@ func TestJoinPublishesFirstReaderImmediately(t *testing.T) { } } -// TestTickFreeze: a second reader joining within the same tick must not change -// the published set (topology stays stable between ticks). -func TestTickFreeze(t *testing.T) { +// New readers are published immediately; renewal does not advance the epoch. +func TestJoinPublishesGrowthImmediately(t *testing.T) { clk := newClock() r := newReg(clk, 5*time.Second, 10*time.Second) - first := r.Join("f1", "A", 0) - - // B joins in the same tick: still frozen at [A]. - resp := r.Join("f1", "B", 0) - if len(resp.Readers) != 1 || resp.Readers[0] != "A" { - t.Errorf("within tick readers = %v, want [A] (frozen)", resp.Readers) + first := r.Join("f1", "B", 0) + next := r.Join("f1", "A", 0) + if !equalStrings(next.Readers, []string{"A", "B"}) || next.EpochS != first.EpochS+1 { + t.Fatal(next) } - if resp.EpochS != first.EpochS { - t.Errorf("epochS changed within a tick: %d -> %d", first.EpochS, resp.EpochS) + renewed := r.Join("f1", "B", 0) + if renewed.EpochS != next.EpochS { + t.Fatal("renewal republished", renewed) } - - // After the tick, both are published and the epoch advances. - clk.advance(6 * time.Second) - resp2 := r.Join("f1", "B", 0) - if len(resp2.Readers) != 2 || resp2.Readers[0] != "A" || resp2.Readers[1] != "B" { - t.Errorf("after tick readers = %v, want [A B]", resp2.Readers) + next.Readers[0] = "mutated" + if got, _ := r.Readers("f1"); !equalStrings(got, []string{"A", "B"}) { + t.Fatal("JOIN response aliases publication", got) } - if resp2.EpochS == first.EpochS { - t.Error("epochS should change when the frozen set changes") +} + +func TestDeparturesWaitForTick(t *testing.T) { + clk := newClock() + r := newReg(clk, 5*time.Second, 10*time.Second) + r.Join("f", "A", 0) + first := r.Join("f", "B", 0) + r.Leave("f", "A") + next := r.Join("f", "C", 0) + if !equalStrings(next.Readers, []string{"A", "B", "C"}) || next.EpochS != first.EpochS+1 { + t.Fatal(next) + } + // Growth must not postpone the existing removal tick. + clk.advance(5 * time.Second) + got, epoch := r.Readers("f") + if !equalStrings(got, []string{"B", "C"}) || epoch != next.EpochS+1 { + t.Fatal(got, epoch) + } + // A rejoining departed reader is immediately visible again. + if got := r.Join("f", "A", 0); !equalStrings(got.Readers, []string{"A", "B", "C"}) { + t.Fatal(got) } }