diff --git a/CHANGELOG.md b/CHANGELOG.md index 3026085..fb7b697 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,33 @@ All notable changes to DART are documented here. This project follows HRW score formula, `ChunkKey`, and epoch framing) is called out explicitly in every release's upgrade notes. +## [Unreleased] + +### Fixes + +- Preserve escaped origin paths in fallback cache identities; encoded slashes + and separators no longer conflate different immutable objects. Digest + recognition respects encoded path boundaries. +- Scope size/Range metadata and probe singleflight to each origin endpoint, + while retaining digest-keyed block deduplication across origins. +- Abort committed peer streams on source errors or known relay geometry + mismatches so a partial chunked response cannot look like a successful block. + Quarantine parents serving invalid block geometry so retries can use origin. + Record transport/validation outcomes with admission-generation permits so + older in-flight completions cannot undo quarantine or consume new probes. + Wait for all admitted half-open probes before recovering, preserving a + sibling protocol failure even when another probe succeeds first. + Decline inconsistent block key/Origin URL pairs before cache or upstream access + so malformed requester addressing cannot quarantine honest parents. + +### Upgrade notes + +Digest keys, ChunkKey, HRW and their golden vectors are unchanged. Fallback +identities for affected non-digest paths now use the escaped path rather than +the decoded path; upgrade all nodes together when serving such URLs. Cache +stores restart cold. Each new endpoint now needs its own one-byte capability +probe even when another endpoint has already populated the digest cache. + ## [v0.2.0] — 2026-08-26 The first release since v0.1.0: **30 merged fix PRs** (#23–#44, #47, #48, diff --git a/docs/adr/0004-preserve-escaped-origin-identities.md b/docs/adr/0004-preserve-escaped-origin-identities.md new file mode 100644 index 0000000..a10217f --- /dev/null +++ b/docs/adr/0004-preserve-escaped-origin-identities.md @@ -0,0 +1,35 @@ +# ADR-0004: Preserve escaped origin identities independently of endpoint probes + +- Status: accepted +- Date: 2026-10-12 +- Supersedes: none +- Binds: docs/chunk.md Stability Contract; docs/fetch.md origin probes; docs/engine.md endpoint metadata + +## Context + +Decoded URL paths collapse distinct immutable origins into the same cache key. +A content digest can identify shared bytes across origins, but cannot establish +that both endpoints support Range or expose the same size metadata. Existing +ADRs preserve trusted, immutable origins; none specifies URL identity encoding. + +Upstream reports: [#77](https://github.com/data-accelerator/dart/issues/77) and +[#78](https://github.com/data-accelerator/dart/issues/78). + +## Decision + +Preserve escaped path boundaries in fallback object identities and keep endpoint +capability probes separate from digest-addressed content deduplication. + +## Consequences + +Affected non-digest paths require coordinated node upgrades; stores restart +cold. Digest identities, ChunkKey construction and HRW golden vectors remain +unchanged. A new endpoint performs its own capability probe even when another +endpoint has already populated the digest cache. + +## Alternatives considered + +Decoding before splitting paths loses encoded slash boundaries. Stripping +separators conflates distinct paths. Digest-keying endpoint probes saves a +request but incorrectly shares endpoint-specific Range capability. Giving up +cross-origin digest block reuse would avoid that coupling at unnecessary cost. diff --git a/docs/adr/README.md b/docs/adr/README.md index 6289cd8..db687bb 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -127,3 +127,4 @@ acceptable in current law — it is a pointer, not a narrative. | [0001](./0001-minimalist-assumptions-as-contract.md) | Minimalist assumptions are contract | accepted | | [0002](./0002-visible-ascii-member-ids.md) | Visible-ASCII member IDs instead of an epoch reframe | accepted | | [0003](./0003-shutdown-safe-join-contract.md) | Shutdown safe-join: never close the store under a live handler | accepted | +| [0004](./0004-preserve-escaped-origin-identities.md) | Preserve escaped origin identities independently of endpoint probes | accepted | diff --git a/docs/chunk.md b/docs/chunk.md index 7f38785..a345c89 100644 --- a/docs/chunk.md +++ b/docs/chunk.md @@ -110,15 +110,17 @@ replicas. **Part of the wire protocol** — see §6. **Field exclusion**: the `0x1F` separator must never appear inside a field, or the serialization is not injective (`("a","b␟c")` would collide with `("a␟b","c")`). Enforced on both sides: a namespace containing `0x1F` is -rejected at engine construction, and derived object identities have `0x1F` -stripped (a URL can smuggle one in via a percent-encoded `%1F` in the path). +rejected at engine construction, and derived object identities retain `0x1F` as an escape (`%1F`), never as a literal +separator. Deleting it would conflate `/a%1Fb` with `/ab`. Parsed paths use +`URL.EscapedPath`; malformed raw fallbacks escape both `%` and `0x1F`. Note that cross-field collisions are the only case: within one namespace the fixed 8-byte chunk-index suffix forces equal objectIDs anyway. `UnitSep` (0x1F) is the field separator `ChunkKey` mixes between fields; it is why `namespace` is rejected at engine construction when it contains 0x1F and -why derived object identities have it stripped (`stripUnitSep`): the separator -must never appear inside a field, or serialization would not be injective. +why derived object identities encode it (`escapeUnitSep` in malformed raw +fallbacks): a literal separator must never appear inside a field, or serialization +would not be injective. ### 3.4.1 `func IsDigest(s string) bool` @@ -135,8 +137,14 @@ Derives the caching identity of a blob URL, preferring content addressing: `sha256:<64 hex>`), returns that digest lower-cased with `contentAddressed=true` (enables cross-origin/registry dedup; marks the object immutable); -- otherwise returns a canonical URL (lower-cased scheme+host, path without - query/fragment) with `contentAddressed=false`. +- otherwise returns a canonical URL (lower-cased scheme+host, **escaped** path + without query/fragment) with `contentAddressed=false`. + +Path segments are split **before** percent-decoding, so an encoded slash does +not expose a digest segment that is absent from the actual origin path. +Fallback identities preserve `/a%2Fb` versus `/a/b` and `%1F` versus `%251F`; +these may name different immutable objects at the origin. Time and space are +O(URL length); returned identities are owned strings and inputs are unchanged. Non-OCI integrity schemes (npm/pypi `sha512-`, etc.) are **not** recognized here; mapping those is a policy-layer concern. @@ -201,6 +209,9 @@ query and fragment are cut at the first `?`/`#` before the identity is derived. `net/url`, `strings`). - `Config` is a small value passed by value. +The identity correction and rollout decision are recorded in +[ADR-0004](./adr/0004-preserve-escaped-origin-identities.md). + ## 6. Determinism / Stability Contract @@ -212,12 +223,18 @@ query and fragment are cut at the first `?`/`#` before the identity is derived. - `ObjectID`'s digest-extraction and URL-canonicalization rules affect cache identity and dedup; changing them changes cache keys and should be versioned deliberately. +- The escaped-path correction changes identities for affected non-digest URLs + (encoded separators, reserved characters and non-ASCII paths). Digest keys, + `ChunkKey`, HRW and their independent golden vectors remain unchanged. + Upgrade the cluster together when serving these URLs: mixed old/new nodes + may interpret the same path differently. Cache stores start cold on restart; + do not retain old URL-keyed entries through an external warm-cache layer. ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **97.9%** of statements. All functions 100% except `isDigest` - 92.9% (one defensive branch uncovered). +- **Coverage**: **99.1%** of statements. Remaining gaps are defensive digest-recognition + branches; escaped origin identities and cache isolation are covered. - **Reproduce**: ```bash @@ -248,6 +265,8 @@ go test ./internal/chunk/ -cover -count=1 | `TestObjectIDCaseNormalization` | digest and host are lower-cased | | `TestObjectID` | digest extraction, lower-casing, query stripping, host:port not a digest, fallbacks | | `TestObjectIDFallbackStripsQueryAndFragment` | the raw fallback cuts query/fragment too: signatures never move identity | +| `TestObjectIDPreservesEscapedPath` | literal identities pin escaped slashes, separators, percent escapes and digest-boundary recognition | +| `TestServeDistinctEscapedOrigins` (engine) | different same-length origin bodies stay distinct through the write-once cache | | `TestObjectIDNeverContainsSeparator` | a %1F-decoded separator never lands in a derived identity | | `TestIsDigest` | digest-shape recognition incl. rejects (bad hex, short, uppercase algo) | diff --git a/docs/engine.md b/docs/engine.md index e988794..0e8cb29 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -44,7 +44,7 @@ back to all-member routing for that lookup (the reader set is soft state). | Term | Meaning | |---|---| | `Engine` | Orchestrator: Size + Serve over cache/origin. | -| object size | Learned once per object via a 1-byte probe, cached in-process. | +| object size | Learned once per origin endpoint via a 1-byte probe, cached in-process. | | `Handler` | `http.Handler` translating HTTP Range ⇄ `Engine.Serve`. | ## 3. Public API @@ -79,10 +79,18 @@ only when `Cluster`, `Peer`, and `SelfID` are all set. ### 3.2 `func (e *Engine) Size(ctx, url string) (int64, error)` Returns the object's total size, probing origin once (a `bytes=0-0` fetch) and -caching the result by object identity (`chunk.ObjectID`). For immutable -content-addressed objects the size is permanently valid. The probe also -records whether the origin honored the Range: a `200` answer marks the object -Range-unsupported (see §3.9). +caching the result by endpoint identity (`fetch.OriginKey`), independently of +the content digest. The same immutable digest can be served by two endpoints +with different Range support or one with an unknown chunked total. Each is +probed separately; refreshed signatures for the same endpoint reuse metadata. +A `200` answer marks only that endpoint Range-unsupported (see §3.9). + +When the fetcher implements `fetch.Prober`, `Size` calls `Probe` so concurrent +metadata flights also stay endpoint-scoped, even when block flights are keyed +by digest. Custom fetchers without `Prober` receive `Fetch(ctx, url, 0, 0)` and +must not coalesce these probes across endpoints. Blocks still use +`chunk.ObjectID` and remain shared across origins. A metadata hit costs O(URL +length) key derivation and a map lookup; a miss adds one ranged GET. A `206` that hides the total (`Content-Range: bytes 0-0/*`) is a **probe failure**, not a size: block geometry is impossible without the total, and @@ -109,9 +117,16 @@ origin and the peer paths: bytes it forwards: relay responses carry no Content-Length (chunked), so a short clean EOF is indistinguishable from success at the transport — the relay compares the streamed length against the geometry before caching. A - mismatched block is still served on (the leaf's own check protects clients) - but never cached; when the object size cannot be resolved the relayed bytes - cannot be validated and are not cached either. + mismatched block fails the relay and is never cached. The peer server aborts + the committed response, so downstream observes truncation even with chunked + framing. The faulty parent is immediately quarantined through its circuit + breaker, so the next request skips it and falls back to origin. Validation + runs inside `Client.StreamValidated` before any success is recorded, under + the peer request's timeout context. Transport and validation results use + the same admission-generation permit, so a pre-quarantine completion cannot + clear a newer circuit or consume its half-open probe slot. + Cooldown/half-open recovery remains unchanged. When the object + size cannot be resolved, relayed bytes cannot be validated and are not cached. This is load-bearing because the block cache is **write-once per key**: a later correct fetch can never overwrite a bad block, only eviction would clear it. So @@ -172,6 +187,16 @@ relay) **before** cache lookup, relay selection, size probe, or origin I/O. Declined, not errored: the malformed index is the requester's fault, and a 500 would charge this healthy relay on the requester's circuit breaker. +**Origin/key consistency is checked at the same boundary.** When a request +supplies `X-DART-Origin`, both sources recompute its expected chunk key from +`(namespace, chunk.ObjectID(url), ChunkOfBlock(blockIndex))`. A different key +is a requester addressing error and is declined before local cache lookup, +parent selection or any origin probe. This prevents an honest parent serving +a requested cached block from being blamed for a different URL's geometry. +Requests without an Origin URL remain valid cache-only reads; a cold one +cannot relay. Identity checking takes O(URL length) time and space, adds no +network traffic, and preserves cut-through streaming. + ### 3.6 `func (e *Engine) PeerStreamSource() peer.StreamSource` The **cut-through** counterpart of `PeerSource`, for a `peer.StreamServer`: @@ -275,9 +300,11 @@ func (e *Engine) ServePassthrough(ctx context.Context, w http.ResponseWriter, r An origin that ignores `Range` cannot be served through the block layer: every per-block fetch would pull the whole object, and a block so fetched could never be safely cached for a *different* range. When the `Size` probe (a `bytes=0-0` -GET) is answered with `200` instead of `206`, the object is marked +GET) is answered with `200` instead of `206`, that origin endpoint is marked Range-unsupported (process-local, never expires — a restart re-probes), and -`Handler` then serves every request for it with `ServePassthrough`: +`Handler` then serves requests for that endpoint with `ServePassthrough`. +Another endpoint for the same digest can still use the cache and P2P: + - the client's `Range` header is forwarded, and the origin's status, entity headers (`Content-Type`/`Content-Length`/`Content-Range`/`ETag`/ @@ -329,6 +356,11 @@ requests; their bytes are counted as both `client` and `origin_in` wire bytes never turned into a whole-object origin GET (issue #52; the fetch-layer arithmetic guard is in docs/fetch.md §3.4). +6. **No requester identity confusion on the relay path**: when an Origin URL + is supplied, its namespace/object/chunk key must match the requested key. + Inconsistent requests yield a clean decline, even if the local cache holds + that key; requester errors never trigger parent fetches or quarantine. + ## 5. Concurrency & Call Permissions - `Engine` is safe for concurrent use: the size cache is mutex-guarded, the @@ -346,7 +378,7 @@ requests; their bytes are counted as both `client` and `origin_in` wire bytes ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **85.4%** of statements. Uncovered lines are mostly rare I/O +- **Coverage**: **88.4%** of statements (Go 1.27.0). Uncovered lines are mostly rare I/O error branches (short block, mid-stream write error). - **Note**: tests use `t.TempDir()` (store files); in the sandbox export `TMPDIR=$PWD/.gotmp`. @@ -367,6 +399,8 @@ go test ./internal/engine/ -cover -count=1 | `TestServeAcrossBlocksAndChunks` | a range crossing block/chunk boundaries is assembled correctly | | `TestServeEndClampAndBadStart` | end clamped to size; start beyond size errors | | `TestSize` | object size probed correctly | +| `TestSizeScopesRangeCapabilityToOrigin` | both probe orders and known/unknown totals stay origin-scoped; refreshed signatures reuse metadata and digest blocks remain shared | +| `TestConcurrentSizeProbesDoNotShareCapabilities` | a blocked Range-blind endpoint does not delay or contaminate an independent endpoint probe | | `TestSizeHiddenTotalIsProbeFailure` | a 206 hiding the total fails loudly, caches nothing, and never fabricates a size | | `TestCacheHitAvoidsRefetch` | repeated identical serve makes zero origin requests | | `TestConcurrentServe` | overlapping concurrent reads correct and race-free | @@ -391,7 +425,10 @@ go test ./internal/engine/ -cover -count=1 | `TestPeerStreamSourceLocalHit` | locally-held block streamed from the store | | `TestPeerStreamSourceRootFetchesOrigin` | tree root satisfies a relay request from origin and caches it | | `TestStreamRelayChainThroughEngines` | 3-node fanout=1 chain over streaming peer servers: bytes intact, every node cached | -| `TestStreamRelayRejectsWrongLengthBlock` | a short clean chunked relay stream is served on but never cached; a correct relay is cached | +| `TestStreamRelayRejectsWrongLengthBlock` | a short clean parent stream makes the relay abort downstream and never cache; a correct relay is cached | +| `TestRelayDeclinesInconsistentKeyURL` | buffered/streaming, warm/cold relays decline unrelated URLs, wrong chunk indices and namespaces without contacting origins/parents or tripping circuits; valid and cache-only reads still work | +| `TestStreamRelayQuarantinesWrongLengthParent` | a clean but invalid parent response opens that parent's circuit; repeated requests fetch origin once and then use the correct cached block | +| `TestStreamRelayPropagatesParentFailure` | an interrupted parent stream remains an unexpected EOF through a real intermediate engine; no partial block is cached | | `TestStreamRelayDoesNotCacheWhenSizeUnresolved` | relayed bytes are not cached when the object size cannot be resolved | | `TestClassOfMatchesPlacement` | owned iff in HRW top-`Replicas`, else borrowed; `Replicas=3` makes all members owners | | `TestClassOfSingleNodeIsOwned` | without P2P everything cached is owned | @@ -452,7 +489,7 @@ go test ./internal/engine/ -cover -count=1 fan-out and prefetch of subsequent blocks are future work. - **Streaming vs buffering**: blocks are fetched whole into memory (bounded by block size) then written; fine for cache hits/`sendfile` later. -- **Size cache lifetime**: sizes are cached for process life with no TTL — +- **Size cache lifetime**: endpoint probe results are cached for process life with no TTL — correct for immutable objects; mutable objects need the policy layer. - **Mid-stream errors**: once headers are sent, a block error can only truncate the body (client detects via Content-Length); prefetching the first block diff --git a/docs/fetch.md b/docs/fetch.md index f092043..c26db4d 100644 --- a/docs/fetch.md +++ b/docs/fetch.md @@ -125,9 +125,38 @@ into a whole-object GET. Anchor: `internal/fetch/fetch.go::FetchBlock`. ```go type Coalescing struct { F Fetcher /* ... */ } func (c *Coalescing) Fetch(ctx, url string, start, end int64) (Range, error) +func (c *Coalescing) Probe(ctx context.Context, url string) (Range, error) func (c *Coalescing) Open(ctx, url string, header http.Header) (*http.Response, error) ``` +`Probe` implements the optional endpoint-metadata interface: + +```go +type Prober interface { Probe(ctx context.Context, url string) (Range, error) } +func OriginKey(rawURL string) string +``` + +`OriginKey` keeps the endpoint's escaped path and lower-cases scheme/host, +excluding userinfo, query and fragment. It derives no digest: two origins or +paths naming the same digest have distinct keys. Malformed input falls back +to the raw URL without query/fragment; the HTTP fetch still validates it. +Key derivation takes O(URL length) time/space and mutates no input. + +`Probe` performs GET `bytes=0-0`, using a separate singleflight group keyed by +`OriginKey`, regardless of `Coalescing.Key`. `RangeIgnored` and an unknown +`Total` describe the endpoint contacted, so they must never be shared across +origins. Different signatures for one endpoint still coalesce; the upstream +request retains its full credential. Probe flights do not join content +flights, including a one-byte block read. `Probe` has `Fetch`'s deadlines, +caller-cancellation and credential-refusal retry guarantees. Probe results +have the same `Range`/error semantics and are not retained after completion; +its shared bytes have the same ownership rules as `Fetch`. + +If the inner fetcher implements `Prober`, both the shared request and an +authorization-refusal retry delegate to its `Probe`, preserving endpoint +isolation through nested coalescing wrappers. Otherwise the inner fetcher +receives `Fetch(ctx, url, 0, 0)` and must keep capability metadata endpoint-scoped. + Wraps a `Fetcher` with singleflight keyed by `(url, start, end)`: concurrent identical fetches share one call to `F`. The **shared origin request runs on a bounded background context** (`MaxFlight`, default 10m): one caller's @@ -269,7 +298,7 @@ bucket), and that changes three things: ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **91.8%** of statements. The remaining uncovered lines are rare +- **Coverage**: **92.4%** of statements (Go 1.27.0). The remaining uncovered lines are rare I/O-error branches (e.g. a mid-body read error) that need fault injection. - **Reproduce**: @@ -299,6 +328,11 @@ go test ./internal/fetch/ -cover -count=1 | `TestFetchBlockClampsTail` | tail block clamped to size-1 | | `TestFetchBlockUnknownSize` | unknown size fetches natural range; Total revealed | | `TestTotalFromContentRange` | Content-Range total parsing incl. `*`/garbage | +| `TestOriginKey` | independent endpoint identities retain path escapes and strip credentials | +| `TestCoalescingProbeSeparatesEndpointsAndContent` | probes for two endpoints and a one-byte content fetch each reach origin independently | +| `TestCoalescingProbeDedupsSignatures` | signed URLs at one endpoint share one probe while retaining credentials on the upstream request | +| `TestCoalescingProbeComposes` | nested digest-keyed wrappers contact both origin endpoints instead of joining an inner content flight | +| `TestCoalescingProbeDelegatesRefusalRetry` | both the shared probe and a renewed-credential retry preserve the inner Prober interface | | `TestCoalescingDedups` | N concurrent identical fetches ⇒ 1 origin call (`-race`) | | `TestCoalescingDistinctKeys` | different url/range ⇒ separate calls | | **`TestCoalescingKeyCollapsesPresignedURLs`** | **20 concurrent callers with distinct signatures for one block cause 1 origin fetch** | diff --git a/docs/peer.md b/docs/peer.md index 6636eb2..da072a6 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -35,6 +35,11 @@ X-DART-Hop: (relay depth, for loop safety) sources decline anything above `chunk.Config.MaxBlockIndex()` with a `404` (issue #52, see docs/engine.md §3.5). `StoreSource` never does range arithmetic with the index, so it needs no such check. +- When `X-DART-Origin` is present, engine sources verify that the wire chunk + key agrees with the URL's object identity, namespace and block's chunk index. + An inconsistent pair is a requester error and yields `404` before cache or + upstream access; it must not quarantine an honest parent. Cache-only requests + without this header remain valid. - `X-DART-Origin` lets a relay-capable Source fetch a block it does not hold (via its own parent/origin); `X-DART-Hop` bounds relay recursion. - `X-DART-Hop` is optional (absent = depth 0) and must be a non-negative @@ -133,6 +138,7 @@ type StreamSource func(ctx context.Context, req BlockRequest, w io.Writer, sizer func StoreStreamSource(s store.Store) StreamSource type StreamServer struct { NodeID string; Src StreamSource } func (c *Client) Stream(ctx, addr string, req BlockRequest, w io.Writer) (n int64, held bool, err error) +func (c *Client) StreamValidated(ctx context.Context, addr string, req BlockRequest, w io.Writer, validate func(context.Context, int64) error) (n int64, held bool, err error) ``` These are the **cut-through** counterparts: a relay copies bytes from its @@ -147,8 +153,28 @@ bytes after ~one block-transfer time rather than depth x block-transfer time). - The server defers the response header until the first byte is written, so a source that declines (`held=false`) or fails early still yields a clean `404`/`500` rather than a truncated `200`. -- A `StreamSource` MUST NOT write anything when returning `held=false`. +- A clean decline (`held=false`, `err=nil`) MUST NOT write anything. A failure + after partial writes returns an error; the server aborts the committed + response using `http.ErrAbortHandler`, which net/http handles without logging + a panic stack. Returning normally would emit a valid chunked terminator and + hide the failure. The same abort applies to incomplete Content-Length + responses. Once all advertised bytes have been flushed, a later source error + cannot be signaled through truncation; sources needing an end-of-stream failure + signal must omit the sizer and use chunked framing. Early failures still + produce the normal error status. Callers invoking `ServeHTTP` + directly must account for this sentinel panic (the normal HTTP server does). - Each write is flushed, pushing bytes downstream promptly. +- `StreamValidated` calls the validator once on a clean `200` body, before + recording success. Nil validation is equivalent to `Stream`; misses and + body/transport errors do not call it. A validator error means a definitive + block protocol violation: the circuit opens and the method returns the byte + count, `held=true`, and a wrapped error. Return nil when validation is + unavailable. The callback runs inline with a timeout-bounded request context, + must honor that context for I/O, and must not retain/mutate caller-owned + stream state concurrently. Time is O(body length) plus validation work; + auxiliary memory is the same bounded copy buffer as `Stream`, plus the + validator's own allocations. The engine validates geometry this way before + admitting relayed bytes to its write-once cache. ```go srv := &peer.StreamServer{NodeID: myID, Src: engine.PeerStreamSource()} @@ -162,7 +188,7 @@ func NewBreaker(BreakerOptions) *Breaker // {FailureThreshold, Cooldown, HalfOp func (b *Breaker) Allow(addr string) bool func (b *Breaker) RecordSuccess(addr string) func (b *Breaker) RecordFailure(addr string) // soft: spends one unit of budget -func (b *Breaker) RecordHardFailure(addr string) // definitive: opens at once +func (b *Breaker) RecordHardFailure(addr string) // definitive dial/protocol failure: opens at once func (b *Breaker) State(addr string) BreakerState // closed | open | half-open (String() prints those words) func (b *Breaker) Healthy(addr string) bool // usable now (no probe reserved) func (b *Breaker) OpenCount() int @@ -176,8 +202,9 @@ immediately with `ErrCircuitOpen` (no dial at all), while still probing so a recovered peer is picked back up without operator action. States: `closed` → (`FailureThreshold` **consecutive** failures) → `open` → (after -`Cooldown`) → `half-open`, admitting `HalfOpenProbes` probes; a probe success closes -the circuit, a probe failure re-opens it and restarts the cooldown. +`Cooldown`) → `half-open`, admitting `HalfOpenProbes` probes. After the first +success, admission pauses until all admitted probes finish. All must succeed to +close the circuit; a probe failure re-opens it and restarts the cooldown. **A `404` is not a failure.** It is a legitimate answer ("I do not hold that block") and blocks legitimately 404 all the time in a distributed cache — counting @@ -225,6 +252,19 @@ made a departed node slow to route around. Opening on one observation stays safe because opening is cheap and **reversible**: the cooldown plus the half-open probe restores a recovered peer with no operator action. +`StreamValidated` records a validated protocol violation as a hard failure, +without first recording transport success. Every `Client` request (Get, Stream, +StreamValidated and FetchRoster) binds its result to the breaker entry and +generation captured when it starts. Opening a circuit or recovering from +half-open invalidates older results. Checking the permit and recording the +outcome occur under the same mutex: an old success cannot clear quarantine or +consume a newer half-open probe slot; an old failure cannot undo recovery. +Evicted entries are not recreated by their old requests' completions. Fresh +roster requests still bypass admission and can restore discovery reachability, +but do not consume data-probe slots or override outstanding probe outcomes. +The public `Allow`/`Record*` API remains an explicit, unscoped reporting API; +callers managing those reports themselves must synchronize their lifecycle. + Defaults: `DefaultFailureThreshold` 5, `DefaultBreakerCooldown` 5s, `DefaultHalfOpenProbes` 1. `Options.Now` injects a clock for tests. The engine additionally uses `Healthy` to route *around* an open peer when choosing a tree @@ -303,8 +343,10 @@ A cluster booting together *always* has nodes dialing peers that are not listeni yet, and a dial failure opens a circuit on the first attempt; gating discovery on that stalls convergence, and with a cyclic seed list it stalls every node at once. This was observed in a live 3-node run. Because outcomes are still recorded, a -successful roster fetch is also what closes a recovered peer's circuit, so discovery -doubles as the health probe that restores the data path. +successful current-generation roster fetch can close a recovered peer's circuit +when no data probes remain outstanding. It never consumes their slots or overrides +their pending outcomes, so discovery also restores reachability without hiding a +data-probe failure. ## 4. Invariants & Guarantees @@ -338,7 +380,7 @@ doubles as the health probe that restores the data path. ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **87.1%** of statements. Uncovered lines are a couple of rare +- **Coverage**: **90.5%** of statements (Go 1.27.0). Uncovered lines are a couple of rare I/O-error branches. - **Note**: tests use `t.TempDir()` (store files); in the sandbox export `TMPDIR=$PWD/.gotmp`. @@ -367,6 +409,10 @@ go test ./internal/peer/ -race -count=1 | `TestStoreStreamSourceRoundtrip` | streaming server/client returns exact block bytes | | `TestStreamServerSetsContentLength` | known-size block is Content-Length framed, not chunked | | `TestStreamServerMissAndErrors` | miss → 404 with nothing written; bad path → 400; non-GET → 405 | +| `TestClientLateFailureCannotUndoRecovery` | an old failed request cannot reopen a circuit recovered by a current half-open probe | +| `TestStreamValidatedUsesRequestContext` | validation receives the bounded peer request context and the streamed byte count | +| `TestClientLateSuccessCannotClearProtocolQuarantine` | deterministic barriers cover stale Get/Stream/roster success during open and half-open states; only a current probe can recover the circuit | +| `TestStreamServerAbortsCommittedErrors` | real HTTP clients detect unexpected EOF after partial source errors with either chunked or Content-Length framing | | `TestStreamServerEarlyErrorGives500` | source failing before any write → clean 500 | | `TestCutThroughPipelines` | **the first half reaches the client while the source still holds the second** (genuinely pipelined, not buffered) | | `TestStreamRelayChain` | two chained stream servers relay bytes intact | @@ -420,3 +466,7 @@ go test ./internal/peer/ -race -count=1 on; membership-skew convergence is a later addition. - **Zero-copy**: the server currently `Write`s a buffer; serving via `sendfile`/`splice` from the store slot is future optimization. + +Half-open regression tests: `TestHalfOpenProbesWaitForSiblingOutcome` covers +concurrent recovery and a sibling hard-failure veto; +`TestHalfOpenRosterSuccessCannotConsumeDataProbe` covers roster slot ownership. diff --git a/docs/registry.md b/docs/registry.md index eb18624..d16be90 100644 --- a/docs/registry.md +++ b/docs/registry.md @@ -269,7 +269,8 @@ nodes (see docs/dart.md). ## 9. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **86.4%** of statements. +- **Coverage**: **90.5%** of statements (Go 1.27.0). Remaining gaps are rare + transport and response-write error branches. - **Reproduce**: ```bash @@ -303,7 +304,7 @@ asserted. | `TestPullOnly` | POST/PUT/PATCH/DELETE → `405` with `Allow`, and never reach the upstream | | `TestNonV2PathRejected` | non-registry paths → `404` | | `TestUpstreamPathPrefixPreserved` | an upstream subpath is prepended | -| **`TestBlobDedupAcrossRegistries`** | **the same digest via two mirrors/repos is fetched once** | +| **`TestBlobDedupAcrossRegistries`** | **the same digest via two mirrors/repos shares cached blocks; each endpoint only adds its own one-byte capability probe** | Authentication (`auth_test.go`, plus the issue-pinning `issue16_test.go` and `issue51_test.go`), against a fake registry that speaks the token diff --git a/internal/chunk/chunk.go b/internal/chunk/chunk.go index 38722a5..a28c3c8 100644 --- a/internal/chunk/chunk.go +++ b/internal/chunk/chunk.go @@ -198,8 +198,8 @@ const ( // It prefers content addressing: if a content digest can be recovered from the // URL path, that digest (lower-cased) is returned and contentAddressed is true — // enabling dedup across origins/registries and marking the object immutable. -// Otherwise a canonical form of the URL (lower-cased scheme+host, path without -// query/fragment) is returned and contentAddressed is false. +// Otherwise a canonical form of the URL (lower-cased scheme+host, escaped path +// without query/fragment) is returned and contentAddressed is false. // // The query is always excluded, which matters beyond tidiness: a presigned // upstream carries a signature there that differs on every request, so including @@ -210,7 +210,13 @@ const ( func ObjectIDLayout(rawURL string, layout DigestLayout) (id string, contentAddressed bool) { u, err := url.Parse(rawURL) if err == nil { - segs := strings.Split(u.Path, "/") + // Split before unescaping: an encoded slash is part of a path + // segment, not a boundary that can expose a digest in a different + // origin object (e.g. /sha256:%2Fmetadata). + segs := strings.Split(u.EscapedPath(), "/") + for i, seg := range segs { + segs[i], _ = url.PathUnescape(seg) + } for _, seg := range segs { if IsDigest(seg) { return strings.ToLower(seg), true @@ -224,7 +230,10 @@ func ObjectIDLayout(rawURL string, layout DigestLayout) (id string, contentAddre host := strings.ToLower(u.Host) scheme := strings.ToLower(u.Scheme) if scheme != "" || host != "" { - return stripUnitSep(scheme + "://" + host + u.Path), false + // Preserve encoded path boundaries: /a%2Fb and /a/b can be + // different origin objects. EscapedPath also retains %1F rather + // than dropping it and conflating /a%1Fb with /ab. + return scheme + "://" + host + u.EscapedPath(), false } } // Unparseable or scheme/host-less: fall back to the raw string with the @@ -239,30 +248,24 @@ func ObjectIDLayout(rawURL string, layout DigestLayout) (id string, contentAddre return strings.ToLower(seg), true } } - return stripUnitSep(raw), false + if err == nil { + return u.EscapedPath(), false + } + return escapeUnitSep(raw), false } // UnitSep is the field separator ChunkKey mixes between namespace and // objectID. It must never appear inside either field, or serialization would // not be injective: ChunkKey("a","b␟c",0) would equal ChunkKey("a␟b","c",0). // Namespaces containing it are rejected at engine construction; derived object -// identities have it stripped (stripUnitSep). +// identities encode it rather than deleting it (escapeUnitSep). const UnitSep = 0x1F -// stripUnitSep removes the ChunkKey field separator from a derived object -// identity. A URL can deliver a literal 0x1F via a percent-encoded "%1F" in -// the path; dropping it keeps ChunkKey injective. (Digest identities are pure -// hex and never contain it.) -func stripUnitSep(s string) string { - if strings.IndexByte(s, UnitSep) < 0 { - return s - } - return strings.Map(func(r rune) rune { - if r == UnitSep { - return -1 - } - return r - }, s) +// escapeUnitSep reversibly encodes separators in the raw fallback. Percent +// must be escaped too, or a literal separator would collide with "%1F". +// Parsed URLs use EscapedPath; this handles malformed input. +func escapeUnitSep(s string) string { + return strings.NewReplacer("%", "%25", "\x1f", "%1F").Replace(s) } // digestHexLen maps a digest algorithm to its hex-encoded length. Only diff --git a/internal/chunk/origin_identity_test.go b/internal/chunk/origin_identity_test.go new file mode 100644 index 0000000..a5c662c --- /dev/null +++ b/internal/chunk/origin_identity_test.go @@ -0,0 +1,25 @@ +package chunk + +import "testing" + +// These identities are literal URL encodings, independent of ChunkKey's hash. +func TestObjectIDPreservesEscapedPath(t *testing.T) { + for _, tc := range []struct{ raw, want string }{ + {"HTTP://Example.COM/a%2Fb", "http://example.com/a%2Fb"}, + {"http://example.com/a/b", "http://example.com/a/b"}, + {"http://example.com/a%1Fb", "http://example.com/a%1Fb"}, + {"http://example.com/ab", "http://example.com/ab"}, + {"http://example.com/a%251Fb", "http://example.com/a%251Fb"}, + {"/a%1Fb?Signature=x#fragment", "/a%1Fb"}, + {"/ab?Signature=x", "/ab"}, + {"http://example.com/sha256:" + sha256 + "%2Fmetadata", "http://example.com/sha256:" + sha256 + "%2Fmetadata"}, + {"bad%escape/a\x1fb", "bad%25escape/a%1Fb"}, + } { + t.Run(tc.raw, func(t *testing.T) { + got, ca := ObjectID(tc.raw) + if ca || got != tc.want { + t.Fatalf("ObjectID(%q) = %q, %v; want %q, false", tc.raw, got, ca, tc.want) + } + }) + } +} diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 9d1c005..2342d16 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -121,15 +121,15 @@ type Engine struct { rs map[string]*readerSet mu sync.Mutex - sizes map[string]sizeMeta // objectID -> probe result + sizes map[string]sizeMeta // origin endpoint -> probe result // opener streams verbatim origin responses for the passthrough fallback // (Range-ignoring origins); nil when the fetcher cannot stream. opener fetch.Opener } -// sizeMeta caches what the size probe learned about an object: its size, and -// whether the origin honors Range at all. A noRange object must be proxied +// sizeMeta caches an origin endpoint's size and Range capability. +// A noRange object must be proxied // verbatim — a per-block fetch would pull the whole object per block. type sizeMeta struct { size int64 @@ -185,9 +185,9 @@ func New(opt Options) (*Engine, error) { } // Size returns the total size of the object at url, probing origin once and -// caching the result by object identity. The probe fetches a single byte and -// reads the total from the response (Content-Range, or Content-Length if the -// origin ignores Range). A probe that reveals a Range-ignoring origin also +// caching the result by fetch.OriginKey (not by digest). The probe fetches a +// single byte and reads the total from Content-Range (or Content-Length if +// the origin ignores Range). A probe revealing a Range-ignoring origin also // marks the object for passthrough (see RangeUnsupported). // // A 206 response that hides the total (`Content-Range: bytes 0-0/*`) is a @@ -197,9 +197,9 @@ func New(opt Options) (*Engine, error) { // Size therefore returns an error and caches nothing, so a corrected origin // recovers without a restart. func (e *Engine) Size(ctx context.Context, url string) (int64, error) { - oid, _ := chunk.ObjectID(url) + origin := fetch.OriginKey(url) e.mu.Lock() - if sm, ok := e.sizes[oid]; ok { + if sm, ok := e.sizes[origin]; ok { e.mu.Unlock() return sm.size, nil } @@ -208,7 +208,14 @@ func (e *Engine) Size(ctx context.Context, url string) (int64, error) { // Probe with a single-byte ranged GET and take the total from Content-Range. // Deliberately not a HEAD: a presigned upstream is signed for GET only (the verb // is inside the signature), so HEAD would return 403. See fetch.Fetcher. - r, err := e.fetcher.Fetch(ctx, url, 0, 0) + var r fetch.Range + var err error + if p, ok := e.fetcher.(fetch.Prober); ok { + // Content-keyed singleflight must not merge endpoint capabilities. + r, err = p.Probe(ctx, url) + } else { + r, err = e.fetcher.Fetch(ctx, url, 0, 0) + } if err != nil { // bytes=0-0 is satisfiable for every non-empty object (RFC 7233), so a // 416 to the probe means the object is empty: cache size 0 and let the @@ -216,7 +223,7 @@ func (e *Engine) Size(ctx context.Context, url string) (int64, error) { var se *fetch.StatusError if errors.As(err, &se) && se.Code == http.StatusRequestedRangeNotSatisfiable { e.mu.Lock() - e.sizes[oid] = sizeMeta{size: 0} + e.sizes[origin] = sizeMeta{size: 0} e.mu.Unlock() return 0, nil } @@ -229,7 +236,7 @@ func (e *Engine) Size(ctx context.Context, url string) (int64, error) { // For a Range-ignoring origin the total may legitimately stay unknown (-1 // on a chunked 200); it is never fabricated — the object is proxied // verbatim via ServePassthrough, which needs no size. - e.sizes[oid] = sizeMeta{size: r.Total, noRange: r.RangeIgnored} + e.sizes[origin] = sizeMeta{size: r.Total, noRange: r.RangeIgnored} e.mu.Unlock() return r.Total, nil } @@ -242,10 +249,10 @@ func (e *Engine) Size(ctx context.Context, url string) (int64, error) { // The mark is process-local and never expires: an origin does not grow Range // support for an existing object, and a restart re-probes anyway. func (e *Engine) RangeUnsupported(url string) bool { - oid, _ := chunk.ObjectID(url) + origin := fetch.OriginKey(url) e.mu.Lock() defer e.mu.Unlock() - return e.sizes[oid].noRange + return e.sizes[origin].noRange } // ErrRangeNotSatisfiable is returned by Serve for a start beyond the object. @@ -306,6 +313,22 @@ func (e *Engine) relayableBlockIndex(b uint64) bool { return b <= uint64(e.cfg.MaxBlockIndex()) } +// relayableRequest checks requester-supplied addressing before cache lookup or +// parent/origin I/O. A key and Origin URL are independent wire fields: comparing +// a held block against an unrelated URL's geometry would blame an honest peer. +// Cache-only requests omit URL and need only the signed-index bound. +func (e *Engine) relayableRequest(req peer.BlockRequest) bool { + if !e.relayableBlockIndex(req.Key.Block) { + return false + } + if req.URL == "" { + return true + } + oid, _ := chunk.ObjectID(req.URL) + want := chunk.ChunkKey(e.ns, oid, e.cfg.ChunkOfBlock(int64(req.Key.Block))) + return req.Key.Chunk == want +} + // block returns the bytes of one block: from the local cache, else (with P2P) // from its parent in the distribution tree, else from origin. Fetched bytes are // cached. hop is the relay depth (0 when serving a local client). @@ -448,9 +471,9 @@ func (e *Engine) fileKey(objectID string) uint64 { // P2P node; a store-only node can use peer.StoreSource instead. func (e *Engine) PeerSource() peer.Source { return func(ctx context.Context, req peer.BlockRequest) ([]byte, bool, error) { - if !e.relayableBlockIndex(req.Key.Block) { + if !e.relayableRequest(req) { e.mx.recordRelay(false) - return nil, false, nil // malformed wire index: decline cheaply + return nil, false, nil // malformed requester addressing: decline cheaply } if data, ok, err := e.store.Get(req.Key); err != nil { return nil, false, err diff --git a/internal/engine/origin_identity_test.go b/internal/engine/origin_identity_test.go new file mode 100644 index 0000000..151c584 --- /dev/null +++ b/internal/engine/origin_identity_test.go @@ -0,0 +1,33 @@ +package engine + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestServeDistinctEscapedOrigins(t *testing.T) { + for _, paths := range [][2]string{{"/a%2Fb", "/a/b"}, {"/a%1Fb", "/ab"}} { + t.Run(paths[0], func(t *testing.T) { + contents := map[string][]byte{paths[0]: []byte("first"), paths[1]: []byte("other")} + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + content, ok := contents[r.URL.EscapedPath()] + if !ok { + http.NotFound(w, r) + return + } + http.ServeContent(w, r, "blob", time.Unix(0, 0), bytes.NewReader(content)) + })) + defer origin.Close() + e := newEngine(t, testCfg()) + for _, path := range paths { + got := serve(t, e, origin.URL+path, 0, 4) + if !bytes.Equal(got, contents[path]) { + t.Fatalf("Serve(%s) = %q, want %q", path, got, contents[path]) + } + } + }) + } +} diff --git a/internal/engine/origin_probe_test.go b/internal/engine/origin_probe_test.go new file mode 100644 index 0000000..35a02c9 --- /dev/null +++ b/internal/engine/origin_probe_test.go @@ -0,0 +1,145 @@ +package engine + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/chunk" + "github.com/data-accelerator/dart/internal/fetch" +) + +func contentKeyedEngine(t *testing.T) *Engine { + t.Helper() + e, err := New(Options{ + Chunk: testCfg(), Store: openStoreAt(t), + Fetcher: &fetch.Coalescing{F: &fetch.HTTPFetcher{}, Key: func(raw string) string { + id, _ := chunk.ObjectID(raw) + return id + }}, + }) + if err != nil { + t.Fatal(err) + } + return e +} + +func TestSizeScopesRangeCapabilityToOrigin(t *testing.T) { + for _, chunked := range []bool{false, true} { + for _, blindFirst := range []bool{false, true} { + name := "known-total" + if chunked { + name = "unknown-total" + } + if blindFirst { + name += "/blind-first" + } else { + name += "/ranged-first" + } + t.Run(name, func(t *testing.T) { + content := blob(16) + var rangedCalls, blindCalls int64 + blind := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt64(&blindCalls, 1) + if chunked { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + } + _, _ = w.Write(content) + })) + defer blind.Close() + ranged := countingOrigin(t, content, &rangedCalls) + path := "/v2/image/blobs/sha256:" + strings.Repeat("a", 64) + urls := []string{ranged.URL + path, blind.URL + path} + if blindFirst { + urls[0], urls[1] = urls[1], urls[0] + } + e := contentKeyedEngine(t) + for _, raw := range urls { + got, err := e.Size(context.Background(), raw+"?Signature=first") + want := int64(len(content)) + if strings.HasPrefix(raw, blind.URL) && chunked { + want = -1 + } + if err != nil || got != want { + t.Fatalf("Size(%s) = %d, %v; want %d", raw, got, err, want) + } + } + if !e.RangeUnsupported(blind.URL+path) || e.RangeUnsupported(ranged.URL+path) { + t.Fatal("Range capability crossed origin endpoints") + } + if atomic.LoadInt64(&blindCalls) != 1 || atomic.LoadInt64(&rangedCalls) != 1 { + t.Fatal("each origin must receive its own probe") + } + // Refreshed signatures share endpoint metadata, not a new probe. + if _, err := e.Size(context.Background(), ranged.URL+path+"?Signature=renewed"); err != nil { + t.Fatal(err) + } + if atomic.LoadInt64(&rangedCalls) != 1 { + t.Fatal("signature refresh caused another endpoint probe") + } + got := serve(t, e, ranged.URL+path, 0, 15) + if !bytes.Equal(got, content) || !e.store.Has(blockKeyFor(ranged.URL+path)) { + t.Fatal("Range-capable origin did not populate the block cache") + } + // A third Range-capable endpoint still probes, but reuses the + // already-cached digest block rather than fetching it again. + var otherCalls int64 + other := countingOrigin(t, content, &otherCalls) + if got := serve(t, e, other.URL+path, 0, 15); !bytes.Equal(got, content) || atomic.LoadInt64(&otherCalls) != 1 { + t.Fatal("per-origin probes broke cross-origin content-cache dedup") + } + }) + } + } +} + +func TestConcurrentSizeProbesDoNotShareCapabilities(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + blind := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(started) + <-release + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + _, _ = io.WriteString(w, "abcd") + })) + defer blind.Close() + defer close(release) + ranged := countingOrigin(t, []byte("abcd"), nil) + path := "/sha256:" + strings.Repeat("a", 64) + e := contentKeyedEngine(t) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := e.Size(ctx, blind.URL+path) + done <- err + }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("first probe never reached its origin") + } + // The first origin is still blocked. A content-keyed probe flight would + // make this independent endpoint wait for it and miss the deadline. + probeCtx, probeCancel := context.WithTimeout(ctx, 500*time.Millisecond) + defer probeCancel() + if size, err := e.Size(probeCtx, ranged.URL+path); err != nil || size != 4 { + t.Fatalf("independent probe = %d, %v; want 4", size, err) + } + // Release the blocked handler; the deferred close also covers failures. + release <- struct{}{} + if err := <-done; err != nil { + t.Fatal(err) + } + if e.RangeUnsupported(ranged.URL+path) || !e.RangeUnsupported(blind.URL+path) { + t.Fatal("concurrent probes mixed endpoint capabilities") + } +} diff --git a/internal/engine/relay_abort_test.go b/internal/engine/relay_abort_test.go new file mode 100644 index 0000000..3f44593 --- /dev/null +++ b/internal/engine/relay_abort_test.go @@ -0,0 +1,88 @@ +package engine + +import ( + "bytes" + "context" + "errors" + "io" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/peer" +) + +func TestStreamRelayPropagatesParentFailure(t *testing.T) { + content := blob(16) + origin := countingOrigin(t, content, nil) + parent := httptest.NewServer(&peer.StreamServer{Src: func(_ context.Context, _ peer.BlockRequest, w io.Writer, _ func(int64)) (int64, bool, error) { + n, err := w.Write(content[:8]) + if err != nil { + return int64(n), true, err + } + return int64(n), true, errors.New("origin stream interrupted") + }}) + defer parent.Close() + e, _ := tailEngine(t, peer.NewClient(), parent, origin.URL) + relay := httptest.NewServer(&peer.StreamServer{Src: e.PeerStreamSource()}) + defer relay.Close() + var body bytes.Buffer + key := blockKeyFor(origin.URL) + n, held, err := peer.NewClient().Stream(context.Background(), strings.TrimPrefix(relay.URL, "http://"), peer.BlockRequest{Key: key, URL: origin.URL}, &body) + if n != 8 || !held || !errors.Is(err, io.ErrUnexpectedEOF) || !bytes.Equal(body.Bytes(), content[:8]) { + t.Fatalf("relay Stream = %d, %v, %v, body=%x; want aborted partial response", n, held, err, body.Bytes()) + } + if e.store.Has(key) { + t.Fatal("interrupted parent stream populated the relay cache") + } +} + +func TestStreamRelayQuarantinesWrongLengthParent(t *testing.T) { + content := blob(16) + var parentCalls, originCalls int64 + origin := countingOrigin(t, content, &originCalls) + parent := httptest.NewServer(&peer.StreamServer{Src: func(_ context.Context, _ peer.BlockRequest, w io.Writer, _ func(int64)) (int64, bool, error) { + atomic.AddInt64(&parentCalls, 1) + n, err := w.Write(content[:8]) + return int64(n), true, err // clean HTTP EOF, wrong block geometry + }}) + defer parent.Close() + client := peer.NewClient() + client.Breaker = peer.NewBreaker(peer.BreakerOptions{ + FailureThreshold: 3, Cooldown: time.Hour, + }) + e, _ := tailEngine(t, client, parent, origin.URL) + relay := httptest.NewServer(&peer.StreamServer{Src: e.PeerStreamSource()}) + defer relay.Close() + key := blockKeyFor(origin.URL) + caller := peer.NewClient() + addr := strings.TrimPrefix(relay.URL, "http://") + parentAddr := strings.TrimPrefix(parent.URL, "http://") + var body bytes.Buffer + _, _, err := caller.Stream(context.Background(), addr, peer.BlockRequest{Key: key, URL: origin.URL}, &body) + if !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("first relay error = %v, want unexpected EOF", err) + } + if state := client.Breaker.State(parentAddr); state != peer.BreakerOpen { + t.Fatalf("invalid parent circuit = %v, want open", state) + } + if e.store.Has(key) { + t.Fatal("invalid parent block entered the cache") + } + beforeFallback := atomic.LoadInt64(&originCalls) + for i := 0; i < 3; i++ { + body.Reset() + n, held, err := caller.Stream(context.Background(), addr, peer.BlockRequest{Key: key, URL: origin.URL}, &body) + if err != nil || !held || n != int64(len(content)) || !bytes.Equal(body.Bytes(), content) { + t.Fatalf("request %d after quarantine = %d, %v, %v; want correct origin bytes", i, n, held, err) + } + } + if atomic.LoadInt64(&parentCalls) != 1 { + t.Fatal("repeated requests contacted the quarantined parent") + } + if atomic.LoadInt64(&originCalls) != beforeFallback+1 || !e.store.Has(key) { + t.Fatal("relay did not fetch the valid origin block once and cache it") + } +} diff --git a/internal/engine/relay_identity_test.go b/internal/engine/relay_identity_test.go new file mode 100644 index 0000000..17947de --- /dev/null +++ b/internal/engine/relay_identity_test.go @@ -0,0 +1,92 @@ +package engine + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/chunk" + "github.com/data-accelerator/dart/internal/peer" + "github.com/data-accelerator/dart/internal/store" +) + +func TestRelayDeclinesInconsistentKeyURL(t *testing.T) { + for _, streaming := range []bool{false, true} { + for _, warm := range []bool{false, true} { + name := "buffered/cold" + if streaming { + name = "stream/cold" + } + if warm { + name = strings.Replace(name, "cold", "warm", 1) + } + t.Run(name, func(t *testing.T) { + content := blob(16) + var parentCalls, originCalls, unrelatedCalls int64 + origin := countingOrigin(t, content, &originCalls) + unrelated := countingOrigin(t, blob(8), &unrelatedCalls) + key := blockKeyFor(origin.URL) + parentStore := openStoreAt(t) + if err := parentStore.Put(key, content); err != nil { + t.Fatal(err) + } + parentSource := &peer.StreamServer{Src: peer.StoreStreamSource(parentStore)} + parent := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt64(&parentCalls, 1) + parentSource.ServeHTTP(w, r) + })) + defer parent.Close() + client := peer.NewClient() + client.Breaker = peer.NewBreaker(peer.BreakerOptions{Cooldown: time.Hour}) + e, _ := tailEngine(t, client, parent, origin.URL) + if warm { + if err := e.store.Put(key, content); err != nil { + t.Fatal(err) + } + } + var handler http.Handler = &peer.Server{Src: e.PeerSource()} + if streaming { + handler = &peer.StreamServer{Src: e.PeerStreamSource()} + } + relay := httptest.NewServer(handler) + defer relay.Close() + caller := peer.NewClient() + caller.Breaker = peer.NewBreaker(peer.BreakerOptions{}) + addr := strings.TrimPrefix(relay.URL, "http://") + wrongBlock := key + wrongBlock.Block = uint64(testCfg().BlocksPerChunk()) + oid, _ := chunk.ObjectID(origin.URL) + wrongNamespace := store.BlockKey{Chunk: chunk.ChunkKey("different", oid, 0)} + for _, req := range []peer.BlockRequest{ + {Key: key, URL: unrelated.URL}, + {Key: wrongBlock, URL: origin.URL}, + {Key: wrongNamespace, URL: origin.URL}, + } { + data, held, err := caller.Get(context.Background(), addr, req) + if err != nil || held || len(data) != 0 { + t.Fatalf("inconsistent request: %d bytes, held=%v, err=%v; want a clean decline", len(data), held, err) + } + } + if atomic.LoadInt64(&parentCalls) != 0 || atomic.LoadInt64(&originCalls) != 0 || atomic.LoadInt64(&unrelatedCalls) != 0 { + t.Fatal("inconsistent request contacted parent or origin") + } + if client.Breaker.State(strings.TrimPrefix(parent.URL, "http://")) != peer.BreakerClosed || caller.Breaker.State(addr) != peer.BreakerClosed { + t.Fatal("requester error quarantined an honest parent or relay") + } + // A consistent request still works, and a cache-only request + // remains valid without an Origin header once that block is held. + for _, raw := range []string{origin.URL, ""} { + data, held, err := caller.Get(context.Background(), addr, peer.BlockRequest{Key: key, URL: raw}) + if err != nil || !held || !bytes.Equal(data, content) { + t.Fatalf("legitimate request: held=%v, err=%v, bytes=%d", held, err, len(data)) + } + } + }) + } + } +} diff --git a/internal/engine/stream.go b/internal/engine/stream.go index a2ba6b4..09235fa 100644 --- a/internal/engine/stream.go +++ b/internal/engine/stream.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "github.com/data-accelerator/dart/internal/chunk" @@ -54,9 +55,9 @@ func asRelayError(err error) error { // rather than depth × block-transfer time. func (e *Engine) PeerStreamSource() peer.StreamSource { return func(ctx context.Context, req peer.BlockRequest, w io.Writer, sizer func(int64)) (int64, bool, error) { - if !e.relayableBlockIndex(req.Key.Block) { + if !e.relayableRequest(req) { e.mx.recordRelay(false) - return 0, false, nil // malformed wire index: decline cheaply + return 0, false, nil // malformed requester addressing: decline cheaply } // Local hit: serve from the store. Get copies the block out under the // store lock, so a concurrent eviction that reuses the slot cannot tear the @@ -131,22 +132,30 @@ func (e *Engine) relayFromParent(ctx context.Context, req peer.BlockRequest, w i } var buf bytes.Buffer up := peer.BlockRequest{Key: req.Key, URL: req.URL, Hop: req.Hop + 1} - n, held, err := e.peer.Stream(ctx, addr, up, io.MultiWriter(w, &buf)) + // Validate before the peer client records success, using the same + // admission-generation permit as the transport outcome. A late success + // from a pre-quarantine request cannot close the parent's newer circuit. + validated := false + validate := func(validationCtx context.Context, n int64) error { + size, serr := e.Size(validationCtx, req.URL) + if serr != nil || size < 0 || e.RangeUnsupported(req.URL) { + return nil // unvalidated bytes may be served on, but never cached + } + want := e.blockLen(size, int64(req.Key.Block)) + if want <= 0 || n != want || int64(buf.Len()) != want { + return fmt.Errorf("engine: relayed block %d has %d bytes, want %d", req.Key.Block, n, want) + } + validated = true + return nil + } + n, held, err := e.peer.StreamValidated(ctx, addr, up, io.MultiWriter(w, &buf), validate) if err != nil || !held { return n, false, err } - // Cache what we relayed (best effort; a store error must not fail the - // relay) — but only after validating the length against the object - // geometry. Relay responses carry no Content-Length, so the transport - // cannot detect a short clean chunked EOF; the relay must enforce what - // the buffered path enforces in block(), or a wrongly-sized block would - // sit in the write-once cache permanently and propagate to every - // downstream hop. A mismatched block is still served on (the leaf's own - // geometry check in block() protects clients, and these bytes came from - // a node we trust) but is never cached. When the size cannot be resolved - // the relayed bytes cannot be validated, so they are not cached either. - size, serr := e.Size(ctx, req.URL) - if serr == nil && n > 0 && int64(buf.Len()) == e.blockLen(size, int64(req.Key.Block)) { + // Cache only a block validated against the object geometry. Streamed + // mismatches abort downstream and quarantine the actual faulty parent; + // unresolved geometry cannot admit bytes to the write-once cache. + if validated { e.putBlock(req.Key, buf.Bytes(), req.Key.Chunk) } return n, true, nil diff --git a/internal/engine/stream_test.go b/internal/engine/stream_test.go index 0c3368b..ca3cb7b 100644 --- a/internal/engine/stream_test.go +++ b/internal/engine/stream_test.go @@ -3,6 +3,7 @@ package engine import ( "bytes" "context" + "errors" "io" "net/http" "net/http/httptest" @@ -136,8 +137,8 @@ func TestStreamRelayChainThroughEngines(t *testing.T) { // caching. Relay responses carry no Content-Length, so a short clean chunked // EOF is indistinguishable from success at the transport; caching such a block // would poison the write-once cache permanently and propagate downstream. The -// relayed bytes are still served on (the leaf's own geometry check protects -// clients), but a wrongly-sized block must never enter the cache. +// downstream response must be aborted and the wrongly-sized block must never +// enter the cache. func TestStreamRelayRejectsWrongLengthBlock(t *testing.T) { content := blob(16) // exactly one block var originCnt int64 @@ -174,11 +175,11 @@ func TestStreamRelayRejectsWrongLengthBlock(t *testing.T) { var buf bytes.Buffer n, held, err := peer.NewClient().Stream(context.Background(), strings.TrimPrefix(srv.URL, "http://"), peer.BlockRequest{Key: key, URL: origin.URL}, &buf) - if err != nil || !held { - t.Fatalf("Stream held=%v err=%v", held, err) + if !errors.Is(err, io.ErrUnexpectedEOF) || !held { + t.Fatalf("Stream held=%v err=%v, want an aborted partial response", held, err) } if n != 8 { - t.Fatalf("downstream received %d bytes, want the 8 the broken parent sent (served on)", n) + t.Fatalf("downstream received %d bytes, want the 8 sent before abort", n) } if st.Has(key) { t.Error("wrongly-sized relayed block was cached (issue #1)") diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index e189fbd..fa39c65 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -107,6 +107,35 @@ type Fetcher interface { Fetch(ctx context.Context, url string, start, end int64) (Range, error) } +// Prober optionally separates endpoint metadata probes from content fetches. +// Engines use it to keep a digest-keyed flight from sharing one origin's Range +// capability with another. Implementations probe with GET bytes=0-0. +type Prober interface { + Probe(ctx context.Context, url string) (Range, error) +} + +// OriginKey identifies an endpoint, not its content: different hosts or +// escaped paths stay distinct even when they name the same digest. Scheme and +// host are lower-cased; userinfo, query and fragment are excluded so credentials +// are excluded from parsed metadata keys and refreshed signatures share a probe. +// Malformed input falls back to the raw string without query or fragment; the +// fetcher rejects it rather than caching probe metadata. +func OriginKey(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil { + if i := strings.IndexAny(rawURL, "?#"); i >= 0 { + return rawURL[:i] + } + return rawURL + } + u.Scheme = strings.ToLower(u.Scheme) + u.Host = strings.ToLower(u.Host) + u.User = nil + u.RawQuery, u.Fragment, u.RawFragment = "", "", "" + u.ForceQuery = false + return u.String() +} + // HTTPFetcher fetches byte ranges over HTTP/HTTPS. type HTTPFetcher struct { // Client is the HTTP client; nil uses http.DefaultClient. @@ -426,10 +455,12 @@ type Coalescing struct { // DefaultMaxFlight. MaxFlight time.Duration g group + probes group } var _ Fetcher = (*Coalescing)(nil) var _ Opener = (*Coalescing)(nil) +var _ Prober = (*Coalescing)(nil) // identity returns the deduplication identity for url. func (c *Coalescing) identity(url string) string { @@ -471,8 +502,29 @@ func (c *Coalescing) maxFlight() time.Duration { // refusal keeps the thundering-herd protection intact on every normal path. func (c *Coalescing) Fetch(ctx context.Context, url string, start, end int64) (Range, error) { key := c.identity(url) + "\x00" + strconv.FormatInt(start, 10) + ":" + strconv.FormatInt(end, 10) + return c.fetch(ctx, key, &c.g, func(ctx context.Context) (Range, error) { + return c.F.Fetch(ctx, url, start, end) + }) +} + +// Probe coalesces GET bytes=0-0 by OriginKey, independently of Key and of +// content flights. Immutable bytes can be shared across endpoints, but a size +// probe's RangeIgnored/Total metadata describes only the endpoint contacted. +// An inner Prober keeps its endpoint boundary on both the shared request and a +// credential-refusal retry. Otherwise the inner Fetcher receives GET bytes=0-0. +// It has Fetch's cancellation, deadline and credential-refusal retry semantics. +func (c *Coalescing) Probe(ctx context.Context, url string) (Range, error) { + return c.fetch(ctx, OriginKey(url), &c.probes, func(ctx context.Context) (Range, error) { + if p, ok := c.F.(Prober); ok { + return p.Probe(ctx, url) + } + return c.F.Fetch(ctx, url, 0, 0) + }) +} + +func (c *Coalescing) fetch(ctx context.Context, key string, g *group, doFetch func(context.Context) (Range, error)) (Range, error) { for { - flight, lead := c.g.acquire(key, c.maxFlight()) + flight, lead := g.acquire(key, c.maxFlight()) if lead { // Exactly one worker goroutine per flight, started by its leader. // It deliberately outlives the leading caller: caller cancellation @@ -483,8 +535,8 @@ func (c *Coalescing) Fetch(ctx context.Context, url string, start, end int64) (R go func() { flightCtx, cancel := context.WithDeadline(context.Background(), flight.end) defer cancel() - r, err := c.F.Fetch(flightCtx, url, start, end) - c.g.finish(key, flight, r, err) + r, err := doFetch(flightCtx) + g.finish(key, flight, r, err) }() } // Wait inline rather than on a per-caller goroutine: a caller that @@ -514,7 +566,7 @@ func (c *Coalescing) Fetch(ctx context.Context, url string, start, end int64) (R // serves this caller alone, so it uses the caller's context and is // skipped entirely when that caller is already gone. if !lead && refused(err) && ctx.Err() == nil { - r, err = c.F.Fetch(ctx, url, start, end) + r, err = doFetch(ctx) r.Coalesced = false // these bytes did cross the network for us } return r, err diff --git a/internal/fetch/probe_test.go b/internal/fetch/probe_test.go new file mode 100644 index 0000000..a49bad4 --- /dev/null +++ b/internal/fetch/probe_test.go @@ -0,0 +1,233 @@ +package fetch + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" +) + +func TestOriginKey(t *testing.T) { + for _, tc := range []struct{ raw, want string }{ + {"HTTP://User:pass@EXAMPLE.com/a%2Fb?Signature=secret#fragment", "http://example.com/a%2Fb"}, + {"http://example.com/a/b", "http://example.com/a/b"}, + {"http://example.com/a%1Fb?Signature=renewed", "http://example.com/a%1Fb"}, + {"http://example.com/a%251Fb", "http://example.com/a%251Fb"}, + {"http://example.com/path?", "http://example.com/path"}, + {"/relative%2Fpath?secret=x#frag", "/relative%2Fpath"}, + {"bad%escape?secret=x#frag", "bad%escape"}, + {"bad%escape", "bad%escape"}, + } { + if got := OriginKey(tc.raw); got != tc.want { + t.Errorf("OriginKey(%q) = %q, want %q", tc.raw, got, tc.want) + } + } +} + +type probeGateFetcher struct { + started chan string + release chan struct{} +} + +func (f *probeGateFetcher) Fetch(ctx context.Context, raw string, start, end int64) (Range, error) { + f.started <- raw + select { + case <-f.release: + return Range{Data: []byte("x"), Total: 4}, nil + case <-ctx.Done(): + return Range{}, ctx.Err() + } +} + +func TestCoalescingProbeSeparatesEndpointsAndContent(t *testing.T) { + f := &probeGateFetcher{started: make(chan string, 3), release: make(chan struct{})} + c := &Coalescing{F: f, Key: func(string) string { return "same-digest" }} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + defer close(f.release) + var wg sync.WaitGroup + for _, raw := range []string{"http://a/blob", "http://b/blob"} { + wg.Add(1) + go func(raw string) { + defer wg.Done() + _, _ = c.Probe(ctx, raw) + }(raw) + } + wg.Add(1) + go func() { + defer wg.Done() + _, _ = c.Fetch(ctx, "http://a/blob", 0, 0) + }() + // All three flights must reach the fetcher before any is released: + // neither another endpoint nor a one-byte content read is a probe match. + for i := 0; i < 3; i++ { + select { + case <-f.started: + case <-ctx.Done(): + t.Fatal("endpoint/content flights were incorrectly coalesced") + } + } + cancel() + wg.Wait() +} + +// waitContext signals when a caller reaches its context-aware flight wait. +// That lets the test release the origin only after the joiner has arrived. +type waitContext struct { + context.Context + waited chan struct{} + once sync.Once +} + +func (c *waitContext) Done() <-chan struct{} { + c.once.Do(func() { close(c.waited) }) + return c.Context.Done() +} + +func TestCoalescingProbeDedupsSignatures(t *testing.T) { + f := &probeGateFetcher{started: make(chan string, 2), release: make(chan struct{})} + c := &Coalescing{F: f, Key: func(string) string { return "unused-for-probes" }} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(f.release) }) } + defer release() + done := make(chan Range, 2) + go func() { + r, _ := c.Probe(ctx, "http://origin/blob?Signature=first") + done <- r + }() + select { + case raw := <-f.started: + if !strings.Contains(raw, "Signature=") { + t.Fatal("the actual upstream request lost its credential") + } + case <-ctx.Done(): + t.Fatal("probe did not start") + } + joiner := &waitContext{Context: ctx, waited: make(chan struct{})} + go func() { + r, _ := c.Probe(joiner, "http://origin/blob?Signature=renewed") + done <- r + }() + select { + case <-joiner.waited: + case <-ctx.Done(): + t.Fatal("joiner did not start") + } + release() + first := <-done + second := <-done + if first.Total != 4 || second.Total != 4 { + t.Fatal("probe did not preserve metadata") + } + if first.Coalesced == second.Coalesced { + t.Fatal("expected one probe leader and one joiner") + } + if len(f.started) != 0 { + t.Fatal("refreshed signatures did not coalesce") + } +} + +func TestCoalescingProbeComposes(t *testing.T) { + f := &probeGateFetcher{started: make(chan string, 2), release: make(chan struct{})} + inner := &Coalescing{F: f, Key: func(string) string { return "same-digest" }} + outer := &Coalescing{F: inner} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + defer close(f.release) + var wg sync.WaitGroup + for _, raw := range []string{"http://a/blob", "http://b/blob"} { + wg.Add(1) + go func(raw string) { + defer wg.Done() + _, _ = outer.Probe(ctx, raw) + }(raw) + } + // Both endpoints must reach origin while neither is released. Calling the + // inner Fetch instead of Probe joins its content-keyed flight and stalls. + seen := make(map[string]bool) + for i := 0; i < 2; i++ { + select { + case raw := <-f.started: + seen[raw] = true + case <-ctx.Done(): + t.Fatal("nested content coalescing merged endpoint probes") + } + } + if len(seen) != 2 { + t.Fatalf("nested probes contacted %v, want both endpoints", seen) + } + cancel() + wg.Wait() +} + +type probeOnlyFetcher struct { + probe func(context.Context, string) (Range, error) +} + +func (f *probeOnlyFetcher) Fetch(context.Context, string, int64, int64) (Range, error) { + return Range{}, errors.New("probe delegated to content fetch") +} + +func (f *probeOnlyFetcher) Probe(ctx context.Context, raw string) (Range, error) { + return f.probe(ctx, raw) +} + +func TestCoalescingProbeDelegatesRefusalRetry(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + defer unblock() + f := &probeOnlyFetcher{probe: func(ctx context.Context, raw string) (Range, error) { + if strings.Contains(raw, "Signature=expired") { + close(started) + select { + case <-release: + return Range{}, &StatusError{Code: 403} + case <-ctx.Done(): + return Range{}, ctx.Err() + } + } + if !strings.Contains(raw, "Signature=renewed") { + return Range{}, errors.New("retry lost the caller's credential") + } + return Range{Data: []byte("x"), Total: 4}, nil + }} + c := &Coalescing{F: f} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + leader := make(chan error, 1) + go func() { + _, err := c.Probe(ctx, "http://origin/blob?Signature=expired") + leader <- err + }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("leader did not delegate to the underlying Prober") + } + joiner := &waitContext{Context: ctx, waited: make(chan struct{})} + done := make(chan struct{}) + var result Range + var resultErr error + go func() { + result, resultErr = c.Probe(joiner, "http://origin/blob?Signature=renewed") + close(done) + }() + select { + case <-joiner.waited: + case <-ctx.Done(): + t.Fatal("renewed credential did not join the endpoint probe") + } + unblock() + if err := <-leader; !refused(err) { + t.Fatalf("expired leader returned %v, want credential refusal", err) + } + <-done + if resultErr != nil || result.Total != 4 || result.Coalesced { + t.Fatalf("delegated retry = %+v, %v; want own successful probe", result, resultErr) + } +} diff --git a/internal/peer/breaker.go b/internal/peer/breaker.go index d967b2b..5f44211 100644 --- a/internal/peer/breaker.go +++ b/internal/peer/breaker.go @@ -62,6 +62,7 @@ type BreakerOptions struct { // Cooldown is how long the circuit stays open before probing again. Cooldown time.Duration // HalfOpenProbes is how many concurrent probes half-open admits. + // After the first success, admission pauses until admitted probes finish. HalfOpenProbes int // Now overrides the clock (tests). Now func() time.Time @@ -81,10 +82,12 @@ type Breaker struct { // breakerEntry is one peer's circuit. type breakerEntry struct { - state BreakerState - failures int // consecutive failures - openedAt time.Time // when the circuit last opened - inFlight int // half-open probes currently outstanding + state BreakerState + generation uint64 // invalidates results from requests admitted before a transition + failures int // consecutive failures + openedAt time.Time // when the circuit last opened + inFlight int // half-open probes currently outstanding + recovering bool // freeze admission while a successful probe awaits siblings // lastActive stamps the last request activity (Allow or a Record* // report). It keys LRU eviction under map pressure: an entry nobody // routes to anymore is stale state for a departed address. Read-only @@ -145,6 +148,7 @@ func (b *Breaker) entryLocked(addr string) *breakerEntry { if e.state == BreakerOpen && b.now().Sub(e.openedAt) >= b.cooldown { e.state = BreakerHalfOpen e.inFlight = 0 + e.recovering = false } return e } @@ -181,87 +185,109 @@ func (b *Breaker) evictStalestLocked() { // the half-open state it has reserved a probe slot, so the caller MUST report the // outcome via RecordSuccess or RecordFailure. func (b *Breaker) Allow(addr string) bool { + _, allowed := b.begin(addr, true) + return allowed +} + +// breakerPermit binds a Client result to the entry and generation at request +// admission. Entry identity also protects against eviction/reuse of an address. +type breakerPermit struct { + breaker *Breaker + entry *breakerEntry + generation uint64 + probe bool // this request owns a half-open probe slot +} + +// begin can bypass admission for discovery while still scoping its result. +func (b *Breaker) begin(addr string, enforce bool) (breakerPermit, bool) { b.mu.Lock() defer b.mu.Unlock() e := b.entryLocked(addr) - e.lastActive = b.now() // a routing attempt is activity, even when rejected - switch e.state { - case BreakerOpen: - return false - case BreakerHalfOpen: - if e.inFlight >= b.probes { - return false + e.lastActive = b.now() + if enforce { + switch e.state { + case BreakerOpen: + return breakerPermit{}, false + case BreakerHalfOpen: + if e.recovering || e.inFlight >= b.probes { + return breakerPermit{}, false + } + e.inFlight++ } - e.inFlight++ - return true - default: - return true } + return breakerPermit{breaker: b, entry: e, generation: e.generation, probe: enforce && e.state == BreakerHalfOpen}, true } -// RecordSuccess reports that addr answered. A success closes the circuit and +// RecordSuccess reports that addr answered. After outstanding probes finish, +// success closes the circuit and // clears the failure count: the peer is demonstrably reachable. func (b *Breaker) RecordSuccess(addr string) { - b.mu.Lock() - defer b.mu.Unlock() - e := b.entryLocked(addr) - e.lastActive = b.now() - if e.inFlight > 0 { - e.inFlight-- - } - e.state = BreakerClosed - e.failures = 0 + b.record(addr, outcomeAnswered, nil) } // RecordFailure reports that addr failed to answer (transport error, timeout, or // an unexpected status). Reaching the threshold opens the circuit; a failure while // half-open re-opens it immediately, restarting the cooldown. func (b *Breaker) RecordFailure(addr string) { - b.mu.Lock() - defer b.mu.Unlock() - e := b.entryLocked(addr) - e.lastActive = b.now() - if e.inFlight > 0 { - e.inFlight-- - } - e.failures++ - switch { - case e.state == BreakerHalfOpen: - // A failed probe re-opens with a fresh cooldown. - e.state = BreakerOpen - e.openedAt = b.now() - case e.state != BreakerOpen && e.failures >= b.threshold: - e.state = BreakerOpen - e.openedAt = b.now() - } - // A failure that arrives while already Open (a late in-flight request dying) - // must NOT restamp openedAt: the cooldown starts when the circuit opens, - // otherwise a steady trickle of late completions pins it open forever. + b.record(addr, outcomeSoftFail, nil) } -// RecordHardFailure reports that addr could not be reached at all (the dial -// failed or timed out). Unlike a slow or erroring peer, that is definitive, so -// the circuit opens immediately instead of spending the whole failure budget -// rediscovering it one dial timeout at a time — which is what made an abruptly -// dead machine take tens of seconds to route around. +// RecordHardFailure reports a definitive peer failure: a failed dial, or a +// validated block protocol violation despite a completed HTTP exchange. It +// opens immediately rather than spending the whole failure budget rediscovering +// an unusable peer one request at a time. // // Opening on a single observation is safe because opening is cheap and // reversible: the cooldown plus the half-open probe brings a recovered peer back // without operator action. func (b *Breaker) RecordHardFailure(addr string) { + b.record(addr, outcomeHardFail, nil) +} + +// record validates the request generation and updates its outcome under one +// lock. An old completion cannot close a newer quarantine, consume a current +// half-open probe, or recreate an entry evicted while that request was flying. +// The public Record* methods retain their explicit, unscoped reporting API; +// Client uses permits so delayed network completions are generation-safe. +func (b *Breaker) record(addr string, o outcome, permit *breakerPermit) { b.mu.Lock() defer b.mu.Unlock() + if permit != nil { + e := b.peers[addr] + if e != permit.entry || e.generation != permit.generation { + return + } + } e := b.entryLocked(addr) e.lastActive = b.now() - if e.inFlight > 0 { + if e.inFlight > 0 && (permit == nil || permit.probe) { e.inFlight-- } + if o == outcomeAnswered { + if e.state == BreakerHalfOpen && e.inFlight > 0 { + // Keep admitted siblings in the current generation until all + // report. A failure must still veto a preceding success. Freeze + // admission so a busy peer cannot grow this cohort forever. + e.recovering = true + return + } + if e.state != BreakerClosed { + e.generation++ + e.inFlight = 0 + } + e.state, e.failures = BreakerClosed, 0 + e.recovering = false + return + } e.failures++ - if e.state != BreakerOpen { + if e.state != BreakerOpen && (o == outcomeHardFail || e.state == BreakerHalfOpen || e.failures >= b.threshold) { e.state = BreakerOpen + e.generation++ + e.inFlight = 0 + e.recovering = false e.openedAt = b.now() } - // Already open: keep the original openedAt (see RecordFailure). + // Already-open failures never extend cooldown; old permits are ignored. } // State returns addr's current circuit state, applying any due transition. diff --git a/internal/peer/half_open_cohort_test.go b/internal/peer/half_open_cohort_test.go new file mode 100644 index 0000000..b2ab627 --- /dev/null +++ b/internal/peer/half_open_cohort_test.go @@ -0,0 +1,139 @@ +package peer + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/store" +) + +func TestHalfOpenProbesWaitForSiblingOutcome(t *testing.T) { + for _, invalidSibling := range []bool{false, true} { + name := "all-success" + if invalidSibling { + name = "hard-failure" + } + t.Run(name, func(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/1") { + close(started) + <-release + if invalidSibling { + _, _ = io.WriteString(w, "!") + return + } + } + _, _ = io.WriteString(w, "abc") + })) + defer srv.Close() + defer unblock() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + clk := newBrkClock() + client := NewClient() + client.Breaker = NewBreaker(BreakerOptions{HalfOpenProbes: 2, Cooldown: time.Minute, Now: clk.now}) + addr := strings.TrimPrefix(srv.URL, "http://") + client.Breaker.RecordHardFailure(addr) + clk.advance(2 * time.Minute) + invalid := errors.New("wrong geometry") + done := make(chan error, 1) + go func() { + _, _, err := client.StreamValidated(ctx, addr, BlockRequest{Key: store.BlockKey{Block: 1}}, io.Discard, + func(_ context.Context, n int64) error { + if n != 3 { + return invalid + } + return nil + }) + done <- err + }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("sibling probe never started") + } + if data, held, err := client.Get(ctx, addr, BlockRequest{Key: store.BlockKey{Block: 2}}); err != nil || !held || string(data) != "abc" { + t.Fatalf("first successful probe = %q, %v, %v", data, held, err) + } + if state := client.Breaker.State(addr); state != BreakerHalfOpen { + t.Fatalf("first success discarded its sibling: state=%v, want half-open", state) + } + // Freeze the existing probe cohort after a success; otherwise new + // probes could keep joining and prevent the cohort from draining. + _, _, err := client.Get(ctx, addr, BlockRequest{Key: store.BlockKey{Block: 3}}) + if !errors.Is(err, ErrCircuitOpen) { + t.Fatalf("new request joined a recovering cohort: %v", err) + } + unblock() + err = <-done + want := BreakerClosed + if invalidSibling { + want = BreakerOpen + if !errors.Is(err, invalid) { + t.Fatalf("sibling protocol result = %v", err) + } + } else if err != nil { + t.Fatal(err) + } + if state := client.Breaker.State(addr); state != want { + t.Fatalf("cohort completed in state=%v, want %v", state, want) + } + }) + } +} + +func TestHalfOpenRosterSuccessCannotConsumeDataProbe(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == RosterPath { + _, _ = io.WriteString(w, `{"epoch":"1","members":[]}`) + return + } + close(started) + <-release + _, _ = io.WriteString(w, "!") + })) + defer srv.Close() + defer unblock() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + clk := newBrkClock() + client := NewClient() + client.Breaker = NewBreaker(BreakerOptions{Cooldown: time.Minute, Now: clk.now}) + addr := strings.TrimPrefix(srv.URL, "http://") + client.Breaker.RecordHardFailure(addr) + clk.advance(2 * time.Minute) + invalid := errors.New("wrong geometry") + done := make(chan error, 1) + go func() { + _, _, err := client.StreamValidated(ctx, addr, BlockRequest{}, io.Discard, func(context.Context, int64) error { return invalid }) + done <- err + }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("data probe never started") + } + if _, _, err := client.FetchRoster(ctx, addr, "", ""); err != nil { + t.Fatal(err) + } + if state := client.Breaker.State(addr); state != BreakerHalfOpen { + t.Fatalf("roster success consumed the data probe: %v", state) + } + unblock() + if err := <-done; !errors.Is(err, invalid) || client.Breaker.State(addr) != BreakerOpen { + t.Fatalf("pending protocol failure was lost: %v, state=%v", err, client.Breaker.State(addr)) + } +} diff --git a/internal/peer/peer.go b/internal/peer/peer.go index d288749..5fc4474 100644 --- a/internal/peer/peer.go +++ b/internal/peer/peer.go @@ -314,32 +314,29 @@ func (c *Client) http() *http.Client { } // allow consults the breaker before dialing addr. -func (c *Client) allow(addr string) error { - if c.Breaker != nil && !c.Breaker.Allow(addr) { - return fmt.Errorf("peer %s: %w", addr, ErrCircuitOpen) +func (c *Client) allow(addr string) (breakerPermit, error) { + if c.Breaker == nil { + return breakerPermit{}, nil + } + permit, allowed := c.Breaker.begin(addr, true) + if !allowed { + return breakerPermit{}, fmt.Errorf("peer %s: %w", addr, ErrCircuitOpen) } - return nil + return permit, nil } -// record reports the outcome of a request to the breaker. -func (c *Client) record(addr string, o outcome) { - if c.Breaker == nil { - return - } - switch o { - case outcomeAnswered: - c.Breaker.RecordSuccess(addr) - case outcomeHardFail: - c.Breaker.RecordHardFailure(addr) - default: - c.Breaker.RecordFailure(addr) +// record reports one request result against its admission generation. +func (c *Client) record(addr string, permit breakerPermit, o outcome) { + if permit.breaker != nil { + permit.breaker.record(addr, o, &permit) } } // 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) { - if err := c.allow(addr); err != nil { + permit, err := c.allow(addr) + if err != nil { return nil, false, err } caller := ctx // caller cancellation is not a peer failure (see below) @@ -347,7 +344,7 @@ func (c *Client) Get(ctx context.Context, addr string, req BlockRequest) (data [ defer cancel() hreq, err := http.NewRequestWithContext(ctx, http.MethodGet, blockURL(addr, req.Key), nil) if err != nil { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return nil, false, err } if req.URL != "" { @@ -365,9 +362,9 @@ func (c *Client) Get(ctx context.Context, addr string, req BlockRequest) (data [ // peer failure; charge nothing. The peer did answer if it got that far, // and recording answered also releases a half-open probe slot. if caller.Err() != nil { - c.record(addr, outcomeAnswered) + c.record(addr, permit, outcomeAnswered) } else { - c.record(addr, classify(err)) + c.record(addr, permit, classify(err)) } return nil, false, err } @@ -377,23 +374,23 @@ func (c *Client) Get(ctx context.Context, addr string, req BlockRequest) (data [ body, err := io.ReadAll(resp.Body) if err != nil { if caller.Err() != nil { - c.record(addr, outcomeAnswered) // we aborted; the peer is fine + c.record(addr, permit, outcomeAnswered) // we aborted; the peer is fine } else { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) } return nil, false, fmt.Errorf("peer %s: read body: %w", addr, err) } - c.record(addr, outcomeAnswered) + c.record(addr, permit, outcomeAnswered) return body, true, nil case http.StatusNotFound: - c.record(addr, outcomeAnswered) // answered; simply does not hold the block + c.record(addr, permit, outcomeAnswered) // answered; simply does not hold the block return nil, false, nil default: if err := upstreamRefusal(addr, resp); err != nil { - c.record(addr, outcomeAnswered) // the peer is fine; our credential is not + c.record(addr, permit, outcomeAnswered) // the peer is fine; our credential is not return nil, false, err } - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return nil, false, fmt.Errorf("peer %s: unexpected status %s", addr, resp.Status) } } diff --git a/internal/peer/roster.go b/internal/peer/roster.go index 1576e83..d03cb39 100644 --- a/internal/peer/roster.go +++ b/internal/peer/roster.go @@ -119,16 +119,20 @@ func (s *RosterServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { // for the cooldown, and with a cyclic seed list it stalls every node at once. // // Outcomes are still *recorded*, which makes this doubly useful: because discovery -// keeps probing at a bounded interval, a peer that comes back has its circuit -// closed by the next successful roster fetch, so the data path recovers without -// waiting for a request to be spent on discovering it. +// keeps probing at a bounded interval, a current-generation success can restore +// reachability without waiting for a block request. Outstanding data probes still +// own their slots and must finish before their circuit can recover. func (c *Client) FetchRoster(ctx context.Context, addr, selfID, selfAddr string) (Roster, string, error) { + var permit breakerPermit + if c.Breaker != nil { + permit, _ = c.Breaker.begin(addr, false) + } ctx, cancel := c.withTimeout(ctx) defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr+RosterPath, nil) if err != nil { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return Roster{}, "", err } if selfID != "" { @@ -140,23 +144,23 @@ func (c *Client) FetchRoster(ctx context.Context, addr, selfID, selfAddr string) resp, err := c.http().Do(req) if err != nil { - c.record(addr, classify(err)) + c.record(addr, permit, classify(err)) return Roster{}, "", fmt.Errorf("peer %s: roster: %w", addr, err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return Roster{}, "", fmt.Errorf("peer %s: roster: unexpected status %s", addr, resp.Status) } body, err := io.ReadAll(io.LimitReader(resp.Body, maxRosterBody)) if err != nil { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return Roster{}, "", fmt.Errorf("peer %s: roster: read: %w", addr, err) } var roster Roster if err := json.Unmarshal(body, &roster); err != nil { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return Roster{}, "", fmt.Errorf("peer %s: roster: decode: %w", addr, err) } @@ -174,7 +178,7 @@ func (c *Client) FetchRoster(ctx context.Context, addr, selfID, selfAddr string) if responder != "" && !rosterHas(roster, responder) { roster.Members = append(roster.Members, RosterMember{ID: responder, Addr: addr, Weight: 1}) } - c.record(addr, outcomeAnswered) + c.record(addr, permit, outcomeAnswered) return roster, responder, nil } diff --git a/internal/peer/stream.go b/internal/peer/stream.go index a162054..acea905 100644 --- a/internal/peer/stream.go +++ b/internal/peer/stream.go @@ -20,10 +20,13 @@ import ( // straight to the downstream socket without buffering the whole block, so a // multi-hop chain pipelines instead of storing-and-forwarding at every hop. // -// Implementations MUST NOT write anything when returning held=false, and MUST -// write exactly n bytes when returning n. If size is known before writing, the -// implementation reports it via the sizer callback so the server can set +// On a clean decline (held=false, err=nil), implementations MUST NOT write +// anything. A failure after partial writes returns err and aborts the response. +// Implementations MUST write exactly n bytes when returning n. If size is +// known before writing, report it via the sizer callback so the server can set // Content-Length; otherwise the response is chunked. +// Once all advertised bytes are flushed, truncation cannot signal a later +// error. Sources with fallible work after the final write must omit the sizer. // // To report that the *origin* refused the fetch-on-behalf (rather than this node // failing), return an error satisfying UpstreamRefusedError so the server can set @@ -102,7 +105,7 @@ func (s *StreamServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Buffer the head of the stream so a "not held" or an immediate error can // still produce a clean 404/500 before any body byte is committed. lw := &lazyWriter{w: w, node: s.NodeID} - n, held, err := s.Src(r.Context(), req, lw, lw.setSize) + _, held, err := s.Src(r.Context(), req, lw, lw.setSize) switch { case err != nil && !lw.started: // Distinguish "the origin refused our requester's credential" from "this @@ -117,14 +120,9 @@ func (s *StreamServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { case !held && !lw.started: http.Error(w, "block not held", http.StatusNotFound) case err != nil: - // Headers already sent; the truncated body signals the failure. When a - // size was reported the client detects the truncation via Content-Length - // mismatch; on a chunked response (no sizer call) an abrupt mid-body - // failure surfaces as unexpected EOF — but a *clean* chunked EOF is - // indistinguishable from success, so a cut-through relay must validate - // the streamed length against the object geometry itself (the engine's - // relay does; see engine.relayFromParent). - _ = n + // Returning normally would emit a clean chunked EOF and hide the + // failure. Abort the response so downstream observes truncation. + panic(http.ErrAbortHandler) } } @@ -168,7 +166,23 @@ func (l *lazyWriter) Write(p []byte) (int, error) { // arrives, without buffering the whole block. It returns the number of bytes // written; held is false if the peer returned 404 (nothing written). func (c *Client) Stream(ctx context.Context, addr string, req BlockRequest, w io.Writer) (n int64, held bool, err error) { - if err := c.allow(addr); err != nil { + return c.stream(ctx, addr, req, w, nil) +} + +// StreamValidated streams like Stream, then validates a clean successful body +// before recording the peer outcome. A validation error is a definitive block +// protocol failure: it opens the parent circuit and returns the streamed byte +// count with held=true and the error. The validator must return nil when it +// cannot validate; it is not called on misses or transport/body errors. Nil +// validation is equivalent to Stream. It runs inline, owns no shared state, +// and must respect the supplied, request-timeout-bounded context for any I/O. +func (c *Client) StreamValidated(ctx context.Context, addr string, req BlockRequest, w io.Writer, validate func(context.Context, int64) error) (n int64, held bool, err error) { + return c.stream(ctx, addr, req, w, validate) +} + +func (c *Client) stream(ctx context.Context, addr string, req BlockRequest, w io.Writer, validate func(context.Context, int64) error) (n int64, held bool, err error) { + permit, err := c.allow(addr) + if err != nil { return 0, false, err } caller := ctx // caller cancellation is not a peer failure (see Get) @@ -176,7 +190,7 @@ func (c *Client) Stream(ctx context.Context, addr string, req BlockRequest, w io defer cancel() hreq, err := http.NewRequestWithContext(ctx, http.MethodGet, blockURL(addr, req.Key), nil) if err != nil { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return 0, false, err } if req.URL != "" { @@ -191,9 +205,9 @@ func (c *Client) Stream(ctx context.Context, addr string, req BlockRequest, w io resp, err := client.Do(hreq) if err != nil { if caller.Err() != nil { - c.record(addr, outcomeAnswered) + c.record(addr, permit, outcomeAnswered) } else { - c.record(addr, classify(err)) + c.record(addr, permit, classify(err)) } return 0, false, err } @@ -203,23 +217,29 @@ func (c *Client) Stream(ctx context.Context, addr string, req BlockRequest, w io n, err = io.Copy(w, resp.Body) if err != nil { if caller.Err() != nil { - c.record(addr, outcomeAnswered) // we aborted; the peer is fine + c.record(addr, permit, outcomeAnswered) // we aborted; the peer is fine } else { - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) } return n, true, fmt.Errorf("peer %s: stream body: %w", addr, err) } - c.record(addr, outcomeAnswered) + if validate != nil { + if err := validate(ctx, n); err != nil { + c.record(addr, permit, outcomeHardFail) + return n, true, fmt.Errorf("peer %s: invalid stream: %w", addr, err) + } + } + c.record(addr, permit, outcomeAnswered) return n, true, nil case http.StatusNotFound: - c.record(addr, outcomeAnswered) // answered; simply does not hold the block + c.record(addr, permit, outcomeAnswered) // answered; simply does not hold the block return 0, false, nil default: if err := upstreamRefusal(addr, resp); err != nil { - c.record(addr, outcomeAnswered) // the peer is fine; our credential is not + c.record(addr, permit, outcomeAnswered) // the peer is fine; our credential is not return 0, false, err } - c.record(addr, outcomeSoftFail) + c.record(addr, permit, outcomeSoftFail) return 0, false, fmt.Errorf("peer %s: unexpected status %s", addr, resp.Status) } } diff --git a/internal/peer/stream_abort_test.go b/internal/peer/stream_abort_test.go new file mode 100644 index 0000000..b6b1922 --- /dev/null +++ b/internal/peer/stream_abort_test.go @@ -0,0 +1,37 @@ +package peer + +import ( + "context" + "errors" + "io" + "net/http/httptest" + "strings" + "testing" +) + +func TestStreamServerAbortsCommittedErrors(t *testing.T) { + for _, sized := range []bool{false, true} { + name := "chunked" + if sized { + name = "content-length" + } + t.Run(name, func(t *testing.T) { + srv := httptest.NewServer(&StreamServer{Src: func(_ context.Context, _ BlockRequest, w io.Writer, sizer func(int64)) (int64, bool, error) { + if sized { + sizer(6) + } + n, err := io.WriteString(w, "abc") + if err != nil { + return int64(n), true, err + } + return int64(n), true, errors.New("source interrupted") + }}) + defer srv.Close() + var body strings.Builder + n, held, err := NewClient().Stream(context.Background(), strings.TrimPrefix(srv.URL, "http://"), BlockRequest{}, &body) + if n != 3 || !held || !errors.Is(err, io.ErrUnexpectedEOF) || body.String() != "abc" { + t.Fatalf("Stream = %d, %v, %v, body=%q; want partial bytes and unexpected EOF", n, held, err, body.String()) + } + }) + } +} diff --git a/internal/peer/stream_generation_test.go b/internal/peer/stream_generation_test.go new file mode 100644 index 0000000..66e0e82 --- /dev/null +++ b/internal/peer/stream_generation_test.go @@ -0,0 +1,191 @@ +package peer + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/data-accelerator/dart/internal/store" +) + +func TestClientLateSuccessCannotClearProtocolQuarantine(t *testing.T) { + for _, mode := range []string{"stream", "get", "roster"} { + for _, afterCooldown := range []bool{false, true} { + name := mode + "/open" + if afterCooldown { + name = mode + "/half-open" + } + t.Run(name, func(t *testing.T) { + oldStarted, probeStarted := make(chan struct{}), make(chan struct{}) + oldRelease, probeRelease := make(chan struct{}), make(chan struct{}) + var oldOnce, probeOnce sync.Once + releaseOld := func() { oldOnce.Do(func() { close(oldRelease) }) } + releaseProbe := func() { probeOnce.Do(func() { close(probeRelease) }) } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == RosterPath || strings.HasSuffix(r.URL.Path, "/1"): + close(oldStarted) + <-oldRelease + case strings.HasSuffix(r.URL.Path, "/2"): + close(probeStarted) + <-probeRelease + default: + _, _ = io.WriteString(w, "!") // clean EOF, invalid geometry + return + } + if r.URL.Path == RosterPath { + _, _ = io.WriteString(w, `{"epoch":"1","members":[]}`) + } else { + _, _ = io.WriteString(w, "abc") + } + })) + defer srv.Close() + defer releaseOld() + defer releaseProbe() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + clk := newBrkClock() + client := NewClient() + client.Breaker = NewBreaker(BreakerOptions{FailureThreshold: 3, Cooldown: time.Minute, Now: clk.now}) + addr := strings.TrimPrefix(srv.URL, "http://") + oldDone := make(chan error, 1) + go func() { + req := BlockRequest{Key: store.BlockKey{Block: 1}} + var err error + switch mode { + case "get": + _, _, err = client.Get(ctx, addr, req) + case "roster": + _, _, err = client.FetchRoster(ctx, addr, "", "") + default: + _, _, err = client.Stream(ctx, addr, req, io.Discard) + } + oldDone <- err + }() + wait := func(ch <-chan struct{}) { + t.Helper() + select { + case <-ch: + case <-ctx.Done(): + t.Fatal("request did not reach its barrier") + } + } + wait(oldStarted) + invalid := errors.New("wrong geometry") + _, _, err := client.StreamValidated(ctx, addr, BlockRequest{}, io.Discard, func(context.Context, int64) error { return invalid }) + if !errors.Is(err, invalid) || client.Breaker.State(addr) != BreakerOpen { + t.Fatalf("invalid stream = %v, state=%v; want a quarantine", err, client.Breaker.State(addr)) + } + wantState := BreakerOpen + probeDone := make(chan error, 1) + if afterCooldown { + clk.advance(2 * time.Minute) + go func() { + _, _, err := client.Stream(ctx, addr, BlockRequest{Key: store.BlockKey{Block: 2}}, io.Discard) + probeDone <- err + }() + wait(probeStarted) + wantState = BreakerHalfOpen + } + // Complete the old successful request only after quarantine (or + // after a new half-open probe is already outstanding). + releaseOld() + if err := <-oldDone; err != nil { + t.Fatal(err) + } + if got := client.Breaker.State(addr); got != wantState { + t.Fatalf("late %s success changed circuit to %v, want %v", mode, got, wantState) + } + _, _, err = client.Stream(ctx, addr, BlockRequest{Key: store.BlockKey{Block: 3}}, io.Discard) + if !errors.Is(err, ErrCircuitOpen) { + t.Fatalf("late completion freed the open/half-open gate: %v", err) + } + if afterCooldown { + // A current-generation successful probe still recovers. + releaseProbe() + if err := <-probeDone; err != nil || client.Breaker.State(addr) != BreakerClosed { + t.Fatalf("current probe did not recover circuit: %v", err) + } + } + }) + } + } +} + +func TestClientLateFailureCannotUndoRecovery(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/1") { + close(started) + <-release + http.Error(w, "old request failed", http.StatusInternalServerError) + return + } + _, _ = io.WriteString(w, "abc") + })) + defer srv.Close() + defer unblock() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + clk := newBrkClock() + client := NewClient() + client.Breaker = NewBreaker(BreakerOptions{FailureThreshold: 1, Cooldown: time.Minute, Now: clk.now}) + addr := strings.TrimPrefix(srv.URL, "http://") + done := make(chan error, 1) + go func() { + _, _, err := client.Get(ctx, addr, BlockRequest{Key: store.BlockKey{Block: 1}}) + done <- err + }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("old request did not start") + } + client.Breaker.RecordHardFailure(addr) + clk.advance(2 * time.Minute) + if _, _, err := client.Stream(ctx, addr, BlockRequest{}, io.Discard); err != nil { + t.Fatal(err) + } + if client.Breaker.State(addr) != BreakerClosed { + t.Fatal("current half-open probe did not recover") + } + unblock() + if err := <-done; err == nil { + t.Fatal("expected the old request's HTTP failure") + } + if state := client.Breaker.State(addr); state != BreakerClosed { + t.Fatalf("old failure reopened recovered circuit: %v", state) + } +} + +func TestStreamValidatedUsesRequestContext(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, "abc") + })) + defer srv.Close() + client := NewClient() + client.Timeout = time.Minute + called := false + n, held, err := client.StreamValidated(context.Background(), strings.TrimPrefix(srv.URL, "http://"), BlockRequest{}, io.Discard, + func(ctx context.Context, n int64) error { + called = true + if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > client.Timeout { + t.Error("validation did not receive the bounded peer request context") + } + if n != 3 { + t.Errorf("validator byte count = %d, want 3", n) + } + return nil + }) + if err != nil || !held || n != 3 || !called { + t.Fatalf("validated stream = %d, %v, %v, called=%v", n, held, err, called) + } +} diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go index ee869b2..f42e20c 100644 --- a/internal/registry/registry_test.go +++ b/internal/registry/registry_test.go @@ -23,14 +23,15 @@ import ( // fakeRegistry is a minimal Registry v2 upstream: one blob addressed by digest, // one mutable manifest, and per-path request counters. type fakeRegistry struct { - blob []byte - digest string - manifest atomic.Value // string, mutable so tag staleness is observable - blobHits atomic.Int64 - manHits atomic.Int64 - pingHits atomic.Int64 - gotHost atomic.Value // string - gotAuth atomic.Value // string + blob []byte + digest string + manifest atomic.Value // string, mutable so tag staleness is observable + blobHits atomic.Int64 + blobProbes atomic.Int64 + manHits atomic.Int64 + pingHits atomic.Int64 + gotHost atomic.Value // string + gotAuth atomic.Value // string } func newFakeRegistry(t *testing.T, blobSize int) (*fakeRegistry, *httptest.Server) { @@ -53,6 +54,9 @@ func newFakeRegistry(t *testing.T, blobSize int) (*fakeRegistry, *httptest.Serve w.WriteHeader(http.StatusOK) case strings.Contains(r.URL.Path, "/blobs/"): fr.blobHits.Add(1) + if r.Header.Get("Range") == "bytes=0-0" { + fr.blobProbes.Add(1) + } // ServeContent gives us correct Range and Content-Length handling, // matching what a real registry (or its CDN) provides. http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(fr.blob)) @@ -459,12 +463,15 @@ func TestBlobDedupAcrossRegistries(t *testing.T) { _, b1 := get(t, m1, pathA, nil) afterFirst := fr.blobHits.Load() + probesAfterFirst := fr.blobProbes.Load() _, b2 := get(t, m2, pathB, nil) if !bytes.Equal(b1, fr.blob) || !bytes.Equal(b2, fr.blob) { t.Fatal("blob bytes mismatch") } - if got := fr.blobHits.Load(); got != afterFirst { - t.Errorf("upstream hit %d more times; the blob was not deduped by digest", got-afterFirst) + // A different endpoint needs its own one-byte capability probe. It must + // not fetch any data blocks: those are still deduplicated by digest. + if got := fr.blobHits.Load(); got != afterFirst+1 || fr.blobProbes.Load() != probesAfterFirst+1 { + t.Errorf("second endpoint made %d requests and %d probes; want only one probe", got-afterFirst, fr.blobProbes.Load()-probesAfterFirst) } }