From 8964a6a415b98b2bf483dd7667066548aea9a24d Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 04:20:45 +0000 Subject: [PATCH 01/11] chunk: preserve escaped origin paths in cache identities Keep encoded path boundaries and separators distinct so different immutable origin objects cannot reuse write-once blocks. Split digest segments before unescaping and document coordinated upgrades for affected URL identities. Add literal identity vectors and real cache-isolation regressions without changing ChunkKey or HRW. --- docs/chunk.md | 27 ++++++++++++---- internal/chunk/chunk.go | 43 +++++++++++++------------ internal/chunk/origin_identity_test.go | 25 ++++++++++++++ internal/engine/origin_identity_test.go | 33 +++++++++++++++++++ 4 files changed, 102 insertions(+), 26 deletions(-) create mode 100644 internal/chunk/origin_identity_test.go create mode 100644 internal/engine/origin_identity_test.go diff --git a/docs/chunk.md b/docs/chunk.md index be01114..1710a0a 100644 --- a/docs/chunk.md +++ b/docs/chunk.md @@ -106,8 +106,9 @@ 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. @@ -126,8 +127,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,12 +208,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 @@ -237,6 +250,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/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/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]) + } + } + }) + } +} From d91d37f54d5c7a5c31668e14882893784216e9cc Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 04:21:22 +0000 Subject: [PATCH 02/11] engine: scope capability probes to origin endpoints Keep endpoint size and Range support separate from digest-addressed content. Add a dedicated coalescing probe group so concurrent origins cannot share metadata flights, while refreshed signatures still coalesce and cached blocks remain shared across registries. Cover both probe orders, unknown totals, independent stalled probes and cross-origin cache reuse. --- docs/engine.md | 30 ++++-- docs/fetch.md | 29 +++++- docs/registry.md | 2 +- internal/engine/engine.go | 33 +++--- internal/engine/origin_probe_test.go | 145 +++++++++++++++++++++++++++ internal/fetch/fetch.go | 45 ++++++++- internal/fetch/probe_test.go | 131 ++++++++++++++++++++++++ internal/registry/registry_test.go | 27 +++-- 8 files changed, 406 insertions(+), 36 deletions(-) create mode 100644 internal/engine/origin_probe_test.go create mode 100644 internal/fetch/probe_test.go diff --git a/docs/engine.md b/docs/engine.md index c75bf05..9497a5b 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 @@ -273,9 +281,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`/ @@ -344,7 +354,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**: **86.9%** of statements. 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`. @@ -365,6 +375,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 | @@ -450,7 +462,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 f0133b0..58ef958 100644 --- a/docs/fetch.md +++ b/docs/fetch.md @@ -118,9 +118,33 @@ into a whole-object GET. ```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`. + 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 @@ -248,7 +272,7 @@ bucket), and that changes three things: ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **93.1%** of statements. The remaining uncovered lines are rare +- **Coverage**: **92.1%** of statements. The remaining uncovered lines are rare I/O-error branches (e.g. a mid-body read error) that need fault injection. - **Reproduce**: @@ -278,6 +302,9 @@ 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 | | `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/registry.md b/docs/registry.md index e382ffb..e4bdceb 100644 --- a/docs/registry.md +++ b/docs/registry.md @@ -264,7 +264,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/engine/engine.go b/internal/engine/engine.go index 9d1c005..37746a6 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. 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/fetch/fetch.go b/internal/fetch/fetch.go index d0d0cc8..f4ea702 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -107,6 +107,33 @@ 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 +// never enter metadata keys and refreshed signatures share an endpoint probe. +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 +453,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 +500,20 @@ 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, url, start, end, key, &c.g) +} + +// 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. +// 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, url, 0, 0, OriginKey(url), &c.probes) +} + +func (c *Coalescing) fetch(ctx context.Context, url string, start, end int64, key string, g *group) (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 @@ -484,7 +525,7 @@ func (c *Coalescing) Fetch(ctx context.Context, url string, start, end int64) (R 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) + g.finish(key, flight, r, err) }() } // Wait inline rather than on a per-caller goroutine: a caller that diff --git a/internal/fetch/probe_test.go b/internal/fetch/probe_test.go new file mode 100644 index 0000000..5700b15 --- /dev/null +++ b/internal/fetch/probe_test.go @@ -0,0 +1,131 @@ +package fetch + +import ( + "context" + "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") + } +} 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) } } From 2ae1c86f5d6d5e5f3c6d2f6566ccff5c7b2e6c35 Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 04:21:43 +0000 Subject: [PATCH 03/11] peer: abort committed streams when relay sources fail Prevent net/http from ending a failed chunked stream as a successful response. Abort after committed source errors, propagate known relay geometry mismatches, and retain clean early failures. Cover both response framing modes and interruption across a real intermediate engine; update protocol behavior and upgrade notes. --- CHANGELOG.md | 20 +++++++++++++++ docs/engine.md | 10 +++++--- docs/peer.md | 11 +++++++-- internal/engine/relay_abort_test.go | 38 +++++++++++++++++++++++++++++ internal/engine/stream.go | 15 ++++++++---- internal/engine/stream_test.go | 11 +++++---- internal/peer/stream.go | 20 ++++++--------- internal/peer/stream_abort_test.go | 37 ++++++++++++++++++++++++++++ 8 files changed, 134 insertions(+), 28 deletions(-) create mode 100644 internal/engine/relay_abort_test.go create mode 100644 internal/peer/stream_abort_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 3026085..77ae9bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,26 @@ 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. + +### 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/engine.md b/docs/engine.md index 9497a5b..cb16d75 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -117,9 +117,10 @@ 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. When the object size cannot be resolved, relayed bytes cannot be + validated and are not cached either. 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 @@ -401,7 +402,8 @@ 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 | +| `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 | diff --git a/docs/peer.md b/docs/peer.md index 3419d0c..e5a5ab2 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -145,7 +145,13 @@ 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 Content-Length responses; 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. ```go @@ -334,7 +340,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**: **89.3%** of statements. 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`. @@ -363,6 +369,7 @@ 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 | +| `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 | diff --git a/internal/engine/relay_abort_test.go b/internal/engine/relay_abort_test.go new file mode 100644 index 0000000..ef4e8ac --- /dev/null +++ b/internal/engine/relay_abort_test.go @@ -0,0 +1,38 @@ +package engine + +import ( + "bytes" + "context" + "errors" + "io" + "net/http/httptest" + "strings" + "testing" + + "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") + } +} diff --git a/internal/engine/stream.go b/internal/engine/stream.go index a2ba6b4..8fe3445 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" @@ -141,12 +142,16 @@ func (e *Engine) relayFromParent(ctx context.Context, req peer.BlockRequest, w i // 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. + // downstream hop. A mismatch fails the stream so the peer server aborts + // its committed response instead of reporting clean chunked success. + // 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)) { + if serr == nil && size >= 0 && !e.RangeUnsupported(req.URL) { + want := e.blockLen(size, int64(req.Key.Block)) + if want <= 0 || n != want || int64(buf.Len()) != want { + return n, false, fmt.Errorf("engine: relayed block %d has %d bytes, want %d", req.Key.Block, n, want) + } 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/peer/stream.go b/internal/peer/stream.go index a162054..0167987 100644 --- a/internal/peer/stream.go +++ b/internal/peer/stream.go @@ -20,9 +20,10 @@ 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. // // To report that the *origin* refused the fetch-on-behalf (rather than this node @@ -102,7 +103,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 +118,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) } } 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()) + } + }) + } +} From a48e85eb4e7f9b2231cd480c71ec9d779cf2799f Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 04:46:36 +0000 Subject: [PATCH 04/11] engine: quarantine parents serving invalid relay geometry Charge definitive block protocol failures to the parent circuit rather than leaving it healthy after a clean HTTP EOF. A soft failure would be reset by every later transport success, so quarantine the invalid parent immediately and retain normal cooldown recovery. Add a real repeated-request regression proving origin fallback and cache reuse. Address Copilot feedback on PR #1; refs #4. --- CHANGELOG.md | 1 + docs/engine.md | 11 +++++-- docs/peer.md | 7 +++- internal/engine/relay_abort_test.go | 50 +++++++++++++++++++++++++++++ internal/engine/stream.go | 7 ++++ internal/peer/breaker.go | 7 ++-- 6 files changed, 76 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 77ae9bb..19d149d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ every release's upgrade notes. 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. ### Upgrade notes diff --git a/docs/engine.md b/docs/engine.md index cb16d75..a2b8150 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -119,8 +119,12 @@ origin and the peer paths: relay compares the streamed length against the geometry before caching. A mismatched block fails the relay and is never cached. The peer server aborts the committed response, so downstream observes truncation even with chunked - framing. When the object size cannot be resolved, relayed bytes cannot be - validated and are not cached either. + framing. The faulty parent is immediately quarantined through its circuit + breaker, so the next request skips it and falls back to origin. A clean HTTP + EOF has already reset transport-level consecutive failures; merely adding a + soft failure afterwards would never accumulate a threshold across repeated + bad blocks. 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 @@ -355,7 +359,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**: **86.9%** of statements. Uncovered lines are mostly rare I/O +- **Coverage**: **87.0%** of statements. 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`. @@ -403,6 +407,7 @@ go test ./internal/engine/ -cover -count=1 | `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 parent stream makes the relay abort downstream and never cache; a correct relay is cached | +| `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 | diff --git a/docs/peer.md b/docs/peer.md index e5a5ab2..f27ef92 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -166,7 +166,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 func (b *Breaker) Healthy(addr string) bool // usable now (no probe reserved) func (b *Breaker) OpenCount() int @@ -229,6 +229,11 @@ 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. +The engine also uses `RecordHardFailure` after validating a definitively invalid +relay block: transport-level success cannot establish that the block is usable. +It quarantines the faulty parent even though the HTTP transfer completed, and +normal cooldown/half-open probes can recover the peer later. + 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 diff --git a/internal/engine/relay_abort_test.go b/internal/engine/relay_abort_test.go index ef4e8ac..3f44593 100644 --- a/internal/engine/relay_abort_test.go +++ b/internal/engine/relay_abort_test.go @@ -7,7 +7,9 @@ import ( "io" "net/http/httptest" "strings" + "sync/atomic" "testing" + "time" "github.com/data-accelerator/dart/internal/peer" ) @@ -36,3 +38,51 @@ func TestStreamRelayPropagatesParentFailure(t *testing.T) { 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/stream.go b/internal/engine/stream.go index 8fe3445..2d50856 100644 --- a/internal/engine/stream.go +++ b/internal/engine/stream.go @@ -150,6 +150,13 @@ func (e *Engine) relayFromParent(ctx context.Context, req peer.BlockRequest, w i if serr == nil && size >= 0 && !e.RangeUnsupported(req.URL) { want := e.blockLen(size, int64(req.Key.Block)) if want <= 0 || n != want || int64(buf.Len()) != want { + if e.peer.Breaker != nil { + // Stream recorded the clean HTTP EOF as success, clearing + // consecutive failures. This block is definitively unusable: + // quarantine its parent immediately, or each next clean but + // invalid response would reset a soft-failure budget again. + e.peer.Breaker.RecordHardFailure(addr) + } return n, false, fmt.Errorf("engine: relayed block %d has %d bytes, want %d", req.Key.Block, n, want) } e.putBlock(req.Key, buf.Bytes(), req.Key.Chunk) diff --git a/internal/peer/breaker.go b/internal/peer/breaker.go index d967b2b..1cd374e 100644 --- a/internal/peer/breaker.go +++ b/internal/peer/breaker.go @@ -239,9 +239,10 @@ func (b *Breaker) RecordFailure(addr string) { // otherwise a steady trickle of late completions pins it open forever. } -// 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 +// RecordHardFailure reports a definitive peer failure: a failed dial, or an +// unusable block whose protocol violation was validated by the engine after +// a transport-level success. The circuit opens immediately rather than +// 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. // From 3a8c1b669b6e44daeb668551414cc2801065d373 Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 05:05:07 +0000 Subject: [PATCH 05/11] peer: scope request outcomes to breaker generations Prevent pre-quarantine transport completions from clearing newer circuits or consuming half-open probe slots. Capture entry identity and generation for block and roster requests, then check and record their outcome under one lock. Validate relay geometry before recording success through StreamValidated, using the bounded request context. Add deterministic concurrent regressions for delayed Get/Stream/roster success, stale failure after recovery and validator context bounds. Address the second Copilot review on PR #1; refs #4. --- CHANGELOG.md | 2 + docs/engine.md | 12 +- docs/peer.md | 32 +++- internal/engine/stream.go | 45 +++--- internal/peer/breaker.go | 115 +++++++------- internal/peer/peer.go | 49 +++--- internal/peer/roster.go | 16 +- internal/peer/stream.go | 42 ++++-- internal/peer/stream_generation_test.go | 191 ++++++++++++++++++++++++ 9 files changed, 376 insertions(+), 128 deletions(-) create mode 100644 internal/peer/stream_generation_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 19d149d..39c91eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ every release's upgrade notes. - 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. ### Upgrade notes diff --git a/docs/engine.md b/docs/engine.md index a2b8150..e001e8f 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -120,10 +120,12 @@ origin and the peer paths: 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. A clean HTTP - EOF has already reset transport-level consecutive failures; merely adding a - soft failure afterwards would never accumulate a threshold across repeated - bad blocks. Cooldown/half-open recovery remains unchanged. When the object + 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 @@ -359,7 +361,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**: **87.0%** of statements. Uncovered lines are mostly rare I/O +- **Coverage**: **86.9%** of statements. 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`. diff --git a/docs/peer.md b/docs/peer.md index f27ef92..14ba1c0 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -131,6 +131,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 @@ -153,6 +154,17 @@ bytes after ~one block-transfer time rather than depth x block-transfer time). 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()} @@ -229,10 +241,17 @@ 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. -The engine also uses `RecordHardFailure` after validating a definitively invalid -relay block: transport-level success cannot establish that the block is usable. -It quarantines the faulty parent even though the HTTP transfer completed, and -normal cooldown/half-open probes can recover the peer later. +`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. +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 @@ -345,7 +364,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**: **89.3%** of statements. Uncovered lines are a couple of rare +- **Coverage**: **89.8%** of statements. 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`. @@ -374,6 +393,9 @@ 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) | diff --git a/internal/engine/stream.go b/internal/engine/stream.go index 2d50856..14e8a11 100644 --- a/internal/engine/stream.go +++ b/internal/engine/stream.go @@ -132,33 +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)) - 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 mismatch fails the stream so the peer server aborts - // its committed response instead of reporting clean chunked success. - // 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 && size >= 0 && !e.RangeUnsupported(req.URL) { + // 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 { - if e.peer.Breaker != nil { - // Stream recorded the clean HTTP EOF as success, clearing - // consecutive failures. This block is definitively unusable: - // quarantine its parent immediately, or each next clean but - // invalid response would reset a soft-failure budget again. - e.peer.Breaker.RecordHardFailure(addr) - } - return n, false, fmt.Errorf("engine: relayed block %d has %d bytes, want %d", req.Key.Block, n, 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 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/peer/breaker.go b/internal/peer/breaker.go index 1cd374e..387142a 100644 --- a/internal/peer/breaker.go +++ b/internal/peer/breaker.go @@ -81,10 +81,11 @@ 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 // 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 @@ -181,88 +182,98 @@ 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 +} + +// 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.inFlight >= b.probes { + return breakerPermit{}, false + } + e.inFlight++ } - e.inFlight++ - return true - default: - return true } + return breakerPermit{breaker: b, entry: e, generation: e.generation}, true } // RecordSuccess reports that addr answered. A 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 a definitive peer failure: a failed dial, or an -// unusable block whose protocol violation was validated by the engine after -// a transport-level success. The circuit opens immediately rather than -// 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 { e.inFlight-- } + if o == outcomeAnswered { + if e.state != BreakerClosed { + e.generation++ + e.inFlight = 0 + } + e.state, e.failures = BreakerClosed, 0 + 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.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/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..8e48863 100644 --- a/internal/peer/roster.go +++ b/internal/peer/roster.go @@ -123,12 +123,16 @@ func (s *RosterServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { // closed by the next successful roster fetch, so the data path recovers without // waiting for a request to be spent on discovering it. 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 0167987..c974797 100644 --- a/internal/peer/stream.go +++ b/internal/peer/stream.go @@ -164,7 +164,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) @@ -172,7 +188,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 != "" { @@ -187,9 +203,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 } @@ -199,23 +215,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_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) + } +} From 5552a7f40ca57b06326f31a32bc8e3f69659cd90 Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 05:23:11 +0000 Subject: [PATCH 06/11] engine: reject inconsistent relay keys and origins Derive the expected chunk key from the supplied Origin URL, namespace and block index before cache lookup or upstream I/O. A malformed requester must not blame an honest parent for an unrelated object geometry. Preserve cache-only reads without Origin and decline caller addressing errors cleanly. Cover buffered/streaming and warm/cold relays, wrong URL/chunk/namespace pairs, healthy circuits and valid control reads. Address the third Copilot review on PR #1; refs #4. --- CHANGELOG.md | 2 + docs/engine.md | 18 ++++- docs/peer.md | 5 ++ internal/engine/engine.go | 20 +++++- internal/engine/relay_identity_test.go | 92 ++++++++++++++++++++++++++ internal/engine/stream.go | 4 +- 6 files changed, 136 insertions(+), 5 deletions(-) create mode 100644 internal/engine/relay_identity_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 39c91eb..84459b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ every release's upgrade notes. 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. + Decline inconsistent block key/Origin URL pairs before cache or upstream access + so malformed requester addressing cannot quarantine honest parents. ### Upgrade notes diff --git a/docs/engine.md b/docs/engine.md index e001e8f..6aa94d8 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -187,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`: @@ -344,6 +354,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 @@ -361,7 +376,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**: **86.9%** of statements. Uncovered lines are mostly rare I/O +- **Coverage**: **87.4%** of statements. 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`. @@ -409,6 +424,7 @@ go test ./internal/engine/ -cover -count=1 | `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 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 | diff --git a/docs/peer.md b/docs/peer.md index 14ba1c0..d8e9f1a 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 diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 37746a6..2342d16 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -313,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). @@ -455,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/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 14e8a11..09235fa 100644 --- a/internal/engine/stream.go +++ b/internal/engine/stream.go @@ -55,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 From d8edd6fd772e62f622f1f9998d6756d99f2d45d7 Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 05:51:11 +0000 Subject: [PATCH 07/11] peer: wait for admitted half-open probe outcomes --- CHANGELOG.md | 2 + docs/peer.md | 12 ++- internal/peer/breaker.go | 22 +++- internal/peer/half_open_cohort_test.go | 139 +++++++++++++++++++++++++ 4 files changed, 168 insertions(+), 7 deletions(-) create mode 100644 internal/peer/half_open_cohort_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 84459b7..fb7b697 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ every release's upgrade notes. 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. diff --git a/docs/peer.md b/docs/peer.md index d8e9f1a..d8643f5 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -197,8 +197,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 @@ -254,7 +255,8 @@ 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. +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. @@ -455,3 +457,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/internal/peer/breaker.go b/internal/peer/breaker.go index 387142a..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 @@ -86,6 +87,7 @@ type breakerEntry struct { 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 @@ -146,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 } @@ -192,6 +195,7 @@ 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. @@ -205,16 +209,17 @@ func (b *Breaker) begin(addr string, enforce bool) (breakerPermit, bool) { case BreakerOpen: return breakerPermit{}, false case BreakerHalfOpen: - if e.inFlight >= b.probes { + if e.recovering || e.inFlight >= b.probes { return breakerPermit{}, false } e.inFlight++ } } - return breakerPermit{breaker: b, entry: e, generation: e.generation}, 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.record(addr, outcomeAnswered, nil) @@ -255,15 +260,23 @@ func (b *Breaker) record(addr string, o outcome, permit *breakerPermit) { } 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++ @@ -271,6 +284,7 @@ func (b *Breaker) record(addr string, o outcome, permit *breakerPermit) { e.state = BreakerOpen e.generation++ e.inFlight = 0 + e.recovering = false e.openedAt = b.now() } // Already-open failures never extend cooldown; old permits are ignored. 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)) + } +} From 0c17e99668bb473c09049e9c5bcf4036f08eeb6c Mon Sep 17 00:00:00 2001 From: Coldwings Soul Date: Mon, 12 Oct 2026 05:52:44 +0000 Subject: [PATCH 08/11] docs: record escaped origin identity rollout decision --- ...0004-preserve-escaped-origin-identities.md | 32 +++++++++++++++++++ docs/adr/README.md | 1 + docs/chunk.md | 3 ++ docs/peer.md | 2 +- 4 files changed, 37 insertions(+), 1 deletion(-) create mode 100644 docs/adr/0004-preserve-escaped-origin-identities.md 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..aefaf07 --- /dev/null +++ b/docs/adr/0004-preserve-escaped-origin-identities.md @@ -0,0 +1,32 @@ +# ADR-0004: Preserve escaped origin identities independently of endpoint probes + +- Status: proposed +- 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. + +## 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..f2298b9 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 | proposed | diff --git a/docs/chunk.md b/docs/chunk.md index 610d29d..dad3c16 100644 --- a/docs/chunk.md +++ b/docs/chunk.md @@ -208,6 +208,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 diff --git a/docs/peer.md b/docs/peer.md index dad9010..9df8f27 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -375,7 +375,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**: **89.8%** of statements. Uncovered lines are a couple of rare +- **Coverage**: **89.9%** of statements. 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`. From 48f3acd18180c0cb54dec264021119469a744fdb Mon Sep 17 00:00:00 2001 From: Coldwings Date: Mon, 12 Oct 2026 14:43:56 +0800 Subject: [PATCH 09/11] docs: align origin repair contracts with current validation Record upstream issue references and Go 1.27 coverage. Clarify that malformed URLs are rejected before metadata caching and roster success cannot consume outstanding data probes. --- docs/adr/0004-preserve-escaped-origin-identities.md | 3 +++ docs/engine.md | 2 +- docs/fetch.md | 2 +- docs/peer.md | 8 +++++--- docs/registry.md | 3 ++- internal/fetch/fetch.go | 4 +++- internal/peer/roster.go | 6 +++--- 7 files changed, 18 insertions(+), 10 deletions(-) diff --git a/docs/adr/0004-preserve-escaped-origin-identities.md b/docs/adr/0004-preserve-escaped-origin-identities.md index aefaf07..db80005 100644 --- a/docs/adr/0004-preserve-escaped-origin-identities.md +++ b/docs/adr/0004-preserve-escaped-origin-identities.md @@ -12,6 +12,9 @@ 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 diff --git a/docs/engine.md b/docs/engine.md index 44defb1..0e8cb29 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -378,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**: **87.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`. diff --git a/docs/fetch.md b/docs/fetch.md index 0a0ec07..3af7e59 100644 --- a/docs/fetch.md +++ b/docs/fetch.md @@ -293,7 +293,7 @@ bucket), and that changes three things: ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **92.1%** of statements. The remaining uncovered lines are rare +- **Coverage**: **92.3%** 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**: diff --git a/docs/peer.md b/docs/peer.md index 9df8f27..14c5a6d 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -340,8 +340,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 @@ -375,7 +377,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**: **89.9%** 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`. diff --git a/docs/registry.md b/docs/registry.md index c2f5c40..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 diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index 7664731..5ba3b11 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -117,7 +117,9 @@ type Prober interface { // 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 -// never enter metadata keys and refreshed signatures share an endpoint probe. +// 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 { diff --git a/internal/peer/roster.go b/internal/peer/roster.go index 8e48863..d03cb39 100644 --- a/internal/peer/roster.go +++ b/internal/peer/roster.go @@ -119,9 +119,9 @@ 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 { From d44e110356cdf6a37b3b473de11eb7fe815b1869 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Mon, 12 Oct 2026 14:55:04 +0800 Subject: [PATCH 10/11] fix(fetch): preserve nested endpoint probe delegation Keep underlying Prober semantics for shared probes and renewed-credential retries so nested content-keyed wrappers cannot merge endpoint capabilities. Add regressions and clarify escaped separator and HTTP stream framing contracts. --- docs/chunk.md | 5 +- docs/fetch.md | 9 +++- docs/peer.md | 7 ++- internal/fetch/fetch.go | 19 +++++-- internal/fetch/probe_test.go | 102 +++++++++++++++++++++++++++++++++++ internal/peer/stream.go | 2 + 6 files changed, 134 insertions(+), 10 deletions(-) diff --git a/docs/chunk.md b/docs/chunk.md index dad3c16..a345c89 100644 --- a/docs/chunk.md +++ b/docs/chunk.md @@ -118,8 +118,9 @@ 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` diff --git a/docs/fetch.md b/docs/fetch.md index 3af7e59..c26db4d 100644 --- a/docs/fetch.md +++ b/docs/fetch.md @@ -152,6 +152,11 @@ 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 @@ -293,7 +298,7 @@ bucket), and that changes three things: ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **92.3%** of statements (Go 1.27.0). 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**: @@ -326,6 +331,8 @@ go test ./internal/fetch/ -cover -count=1 | `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 14c5a6d..da072a6 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -157,8 +157,11 @@ bytes after ~one block-transfer time rather than depth x block-transfer time). 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 Content-Length responses; early - failures still produce the normal error status. Callers invoking `ServeHTTP` + 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 diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index 5ba3b11..fa39c65 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -502,18 +502,27 @@ 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, url, start, end, key, &c.g) + 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, url, 0, 0, OriginKey(url), &c.probes) + 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, url string, start, end int64, key string, g *group) (Range, error) { +func (c *Coalescing) fetch(ctx context.Context, key string, g *group, doFetch func(context.Context) (Range, error)) (Range, error) { for { flight, lead := g.acquire(key, c.maxFlight()) if lead { @@ -526,7 +535,7 @@ func (c *Coalescing) fetch(ctx context.Context, url string, start, end int64, ke go func() { flightCtx, cancel := context.WithDeadline(context.Background(), flight.end) defer cancel() - r, err := c.F.Fetch(flightCtx, url, start, end) + r, err := doFetch(flightCtx) g.finish(key, flight, r, err) }() } @@ -557,7 +566,7 @@ func (c *Coalescing) fetch(ctx context.Context, url string, start, end int64, ke // 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 index 5700b15..a49bad4 100644 --- a/internal/fetch/probe_test.go +++ b/internal/fetch/probe_test.go @@ -2,6 +2,7 @@ package fetch import ( "context" + "errors" "strings" "sync" "testing" @@ -129,3 +130,104 @@ func TestCoalescingProbeDedupsSignatures(t *testing.T) { 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/stream.go b/internal/peer/stream.go index c974797..acea905 100644 --- a/internal/peer/stream.go +++ b/internal/peer/stream.go @@ -25,6 +25,8 @@ import ( // 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 From 89ca6e99d2236b2fe7a3e76eeedb4d613cfdeb53 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Mon, 12 Oct 2026 15:00:05 +0800 Subject: [PATCH 11/11] docs: accept origin identity ADR with reviewed implementation Mark ADR-0004 accepted in the merge-bound revision so the decision and its implementation enter main together after review. --- docs/adr/0004-preserve-escaped-origin-identities.md | 2 +- docs/adr/README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/adr/0004-preserve-escaped-origin-identities.md b/docs/adr/0004-preserve-escaped-origin-identities.md index db80005..a10217f 100644 --- a/docs/adr/0004-preserve-escaped-origin-identities.md +++ b/docs/adr/0004-preserve-escaped-origin-identities.md @@ -1,6 +1,6 @@ # ADR-0004: Preserve escaped origin identities independently of endpoint probes -- Status: proposed +- Status: accepted - Date: 2026-10-12 - Supersedes: none - Binds: docs/chunk.md Stability Contract; docs/fetch.md origin probes; docs/engine.md endpoint metadata diff --git a/docs/adr/README.md b/docs/adr/README.md index f2298b9..db687bb 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -127,4 +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 | proposed | +| [0004](./0004-preserve-escaped-origin-identities.md) | Preserve escaped origin identities independently of endpoint probes | accepted |