diff --git a/docs/engine.md b/docs/engine.md index ec08f0c..1c023dc 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -360,6 +360,7 @@ go test ./internal/engine/ -cover -count=1 | `TestEngineP2PPeerMissFallsBackToOrigin` | empty peer (404) → origin fallback, correct bytes | | `TestEngineP2PSelfOwnerUsesOrigin` | self is owner → peers not consulted, origin used | | `TestPeerSourceRelayFetchOnBehalf` | relay Source fetches-on-behalf and serves a block it did not hold | +| **`TestRelayHopBoundary`** | **both peer-facing sources: hop == maxHop-1 still relays; hop == maxHop declines without touching the origin** | | `TestTreeMultiHopRelay` | 3-node fanout=1 chain: a tail request populates every node via the relay chain | | `TestPeerStreamSourceLocalHit` | locally-held block streamed from the store | | `TestPeerStreamSourceRootFetchesOrigin` | tree root satisfies a relay request from origin and caches it | diff --git a/docs/peer.md b/docs/peer.md index 8d31556..c2aef36 100644 --- a/docs/peer.md +++ b/docs/peer.md @@ -31,10 +31,19 @@ X-DART-Hop: (relay depth, for loop safety) `store.BlockKey.Block` in base-10. - `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 + decimal int; any other value — malformed, negative, or out of range — is + rejected with `400` before the Source is invoked. Accepting a negative hop + would weaken the engine's relay-loop bound (`hop >= maxHop`, incremented at + each relay): a negative start would delay the cutoff, a huge negative one + effectively forever. The bound itself (`maxHop`) is enforced by the engine, + not here, so any non-negative value — including one at or past the bound — + is a valid wire value at this layer. - Responses: `200` + block bytes (with `X-DART-Node`; `Content-Length` when the source knows the size up front, chunked otherwise — e.g. on the cut-through relay path), `404` if the peer cannot provide the block, `400` - for a malformed path, `405` for non-GET, `500` on a source error, and `502` + + for a malformed path or an invalid `X-DART-Hop`, `405` for non-GET, `500` + on a source error, and `502` + `X-DART-Upstream-Status: ` when a relay's origin fetch was refused (§3.6 — the peer is fine; the caller's credential is not). - The path has no embedded URLs, so it is parsed from `URL.Path` (no `//` trap). @@ -330,6 +339,8 @@ go test ./internal/peer/ -race -count=1 | `TestClientMiss` | a not-held block yields `held=false, err=nil` (404) | | `TestServerNodeHeader` | `X-DART-Node` echoed on responses | | `TestServerBadPathAndMethod` | malformed paths → 400/404; non-GET → 405 | +| **`TestServerHopValidation`** | **both servers: absent hop = 0; non-negative hops (incl. ≥ the engine's relay bound) accepted; negative/malformed/overflow hops → 400 and the Source is never invoked** | +| `TestParseHop` | decoder edges: empty, leading plus, `-0`, overflow | | `TestServerSourceError` | a source error → 500 | | `TestClientConnError` | a closed peer port surfaces a transport error | | `TestParseBlockPath` | path parsing incl. max uint64 and rejects | diff --git a/internal/engine/hop_bound_test.go b/internal/engine/hop_bound_test.go new file mode 100644 index 0000000..aac8173 --- /dev/null +++ b/internal/engine/hop_bound_test.go @@ -0,0 +1,88 @@ +package engine + +import ( + "bytes" + "context" + "strings" + "sync/atomic" + "testing" + + "github.com/data-accelerator/dart/internal/chunk" + "github.com/data-accelerator/dart/internal/cluster" + "github.com/data-accelerator/dart/internal/peer" + "github.com/data-accelerator/dart/internal/store" +) + +// TestRelayHopBoundary pins the maxHop gate on both peer-facing sources +// (PeerSource and PeerStreamSource): at hop == maxHop-1 a miss still relays +// (falls through to origin here, this node being a lone root), while at +// hop == maxHop the request is declined without the origin being touched at +// all. Together with the peer transport rejecting negative/malformed hops +// (internal/peer), this is the whole loop-safety contract: hop enters the +// engine in [0, maxHop), grows by one per relay, and is cut off at maxHop. +func TestRelayHopBoundary(t *testing.T) { + content := blob(100) + + cases := []struct { + name string + hop int + wantHeld bool + wantFetch bool + }{ + {"one below the bound relays", maxHop - 1, true, true}, + {"at the bound declines", maxHop, false, false}, + } + + sources := map[string]func(e *Engine, req peer.BlockRequest, buf *bytes.Buffer) (bool, error){ + "buffered": func(e *Engine, req peer.BlockRequest, buf *bytes.Buffer) (bool, error) { + data, held, err := e.PeerSource()(context.Background(), req) + if held { + buf.Write(data) + } + return held, err + }, + "stream": func(e *Engine, req peer.BlockRequest, buf *bytes.Buffer) (bool, error) { + _, held, err := e.PeerStreamSource()(context.Background(), req, buf, func(int64) {}) + return held, err + }, + } + + for srcName, call := range sources { + for _, tc := range cases { + t.Run(srcName+"/"+tc.name, func(t *testing.T) { + var originHits int64 + origin := countingOrigin(t, content, &originHits) + + prov := cluster.NewStaticProvider( + cluster.Member{ID: "R", Addr: "127.0.0.1:1", Weight: 1, State: cluster.Ready}) + e, err := New(Options{ + Chunk: testCfg(), Store: openStoreAt(t), Fetcher: newFetcher(), + Cluster: prov, Peer: peer.NewClient(), SelfID: "R", + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + url := origin.URL + "/blob" + oid, _ := chunk.ObjectID(url) + key := store.BlockKey{Chunk: chunk.ChunkKey("dart", oid, 0), Block: 0} + req := peer.BlockRequest{Key: key, URL: url, Hop: tc.hop} + + var buf bytes.Buffer + held, err := call(e, req, &buf) + if err != nil { + t.Fatalf("source error: %v", err) + } + if held != tc.wantHeld { + t.Errorf("held = %v, want %v", held, tc.wantHeld) + } + if gotFetch := atomic.LoadInt64(&originHits) > 0; gotFetch != tc.wantFetch { + t.Errorf("origin contacted = %v, want %v", gotFetch, tc.wantFetch) + } + if tc.wantHeld && !bytes.Equal(buf.Bytes(), content[:16]) { + t.Errorf("served bytes mismatch: got %d bytes %q", buf.Len(), strings.TrimSpace(buf.String())) + } + }) + } + } +} diff --git a/internal/peer/hop_test.go b/internal/peer/hop_test.go new file mode 100644 index 0000000..8ae0c71 --- /dev/null +++ b/internal/peer/hop_test.go @@ -0,0 +1,120 @@ +package peer + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "testing" +) + +// TestServerHopValidation pins the X-DART-Hop domain on both peer server +// implementations (buffered Server and cut-through StreamServer): the header is +// optional (absent = depth 0) and any non-negative decimal is accepted — the +// relay-depth bound itself is enforced by the engine, not the transport — but a +// malformed, negative, or out-of-range value is rejected with 400 before the +// source is ever invoked. A negative hop used to sail through: the engine only +// declines at hop >= maxHop and every relay increments the value, so a negative +// start delayed the loop-safety cutoff (a huge negative one effectively +// forever) exactly when membership skew creates a relay cycle. +func TestServerHopValidation(t *testing.T) { + blockPath := "/peer/v1/block/abcdef/7" + + servers := map[string]func(record func(hop int)) http.Handler{ + "buffered": func(record func(hop int)) http.Handler { + return &Server{NodeID: "n", Src: func(_ context.Context, req BlockRequest) ([]byte, bool, error) { + record(req.Hop) + return []byte("data"), true, nil + }} + }, + "stream": func(record func(hop int)) http.Handler { + return &StreamServer{NodeID: "n", Src: func(_ context.Context, req BlockRequest, w io.Writer, sizer func(int64)) (int64, bool, error) { + record(req.Hop) + sizer(4) + n, err := w.Write([]byte("data")) + return int64(n), true, err + }} + }, + } + + cases := []struct { + name string + hop string // header value; set=false means the header is absent + set bool + wantStatus int + wantHop int // hop the source must observe (valid cases only) + }{ + {"absent means depth zero", "", false, http.StatusOK, 0}, + {"zero", "0", true, http.StatusOK, 0}, + {"below relay bound", "63", true, http.StatusOK, 63}, + // The transport accepts any non-negative depth; declining at the relay + // bound (maxHop) is the engine's job, so 64 and beyond are still valid + // wire values here. + {"at relay bound", "64", true, http.StatusOK, 64}, + {"large", "4096", true, http.StatusOK, 4096}, + {"negative one", "-1", true, http.StatusBadRequest, 0}, + {"negative large", "-4096", true, http.StatusBadRequest, 0}, + {"min int64", "-9223372036854775808", true, http.StatusBadRequest, 0}, + {"overflow", "9223372036854775808", true, http.StatusBadRequest, 0}, + {"non-numeric", "abc", true, http.StatusBadRequest, 0}, + {"float", "1.5", true, http.StatusBadRequest, 0}, + {"leading space", " 1", true, http.StatusBadRequest, 0}, + {"grouped digits", "1_000", true, http.StatusBadRequest, 0}, + } + + for srvName, build := range servers { + for _, tc := range cases { + t.Run(srvName+"/"+tc.name, func(t *testing.T) { + var calls int + var gotHop int + h := build(func(hop int) { calls++; gotHop = hop }) + + r := httptest.NewRequest(http.MethodGet, blockPath, nil) + if tc.set { + r.Header.Set(HeaderHop, tc.hop) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, r) + + if rec.Code != tc.wantStatus { + t.Fatalf("status = %d, want %d (body %q)", rec.Code, tc.wantStatus, rec.Body.String()) + } + if tc.wantStatus == http.StatusOK { + if calls != 1 { + t.Fatalf("source called %d times, want 1", calls) + } + if gotHop != tc.wantHop { + t.Errorf("source observed hop %d, want %d", gotHop, tc.wantHop) + } + } else if calls != 0 { + t.Errorf("source called %d times on a rejected request, want 0", calls) + } + }) + } + } +} + +// TestParseHop exercises the decoder directly, including the values strconv +// accepts that the wire never produces (a leading plus). +func TestParseHop(t *testing.T) { + cases := []struct { + in string + wantHop int + wantOK bool + }{ + {"", 0, true}, + {"0", 0, true}, + {"7", 7, true}, + {"+5", 5, true}, // strconv.Atoi semantics; harmless and accepted + {"-1", 0, false}, + {"-0", 0, true}, // Atoi("-0") == 0, which is in domain + {"abc", 0, false}, + {"9223372036854775808", 0, false}, + } + for _, tc := range cases { + hop, ok := parseHop(tc.in) + if ok != tc.wantOK || (ok && hop != tc.wantHop) { + t.Errorf("parseHop(%q) = (%d, %v), want (%d, %v)", tc.in, hop, ok, tc.wantHop, tc.wantOK) + } + } +} diff --git a/internal/peer/peer.go b/internal/peer/peer.go index 48f57f2..d288749 100644 --- a/internal/peer/peer.go +++ b/internal/peer/peer.go @@ -62,7 +62,8 @@ type BlockRequest struct { // (the peer should serve only what it already holds). URL string // Hop is the relay depth (X-DART-Hop); incremented at each relay to bound - // recursion and detect loops. + // recursion and detect loops. Always non-negative on the wire: the servers + // reject a malformed or negative header with 400 before calling the Source. Hop int } @@ -100,7 +101,11 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.Error(w, "bad block path", http.StatusBadRequest) return } - hop, _ := strconv.Atoi(r.Header.Get(HeaderHop)) + hop, ok := parseHop(r.Header.Get(HeaderHop)) + if !ok { + http.Error(w, "bad "+HeaderHop+" header", http.StatusBadRequest) + return + } req := BlockRequest{Key: key, URL: r.Header.Get(HeaderOrigin), Hop: hop} data, held, err := s.Src(r.Context(), req) @@ -128,6 +133,24 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { _, _ = w.Write(data) } +// parseHop decodes the X-DART-Hop header. The header is optional: absent means +// depth 0 (a direct, non-relayed request). Any value outside the valid +// domain — malformed, negative, or out of int range — is rejected: the engine +// bounds relay recursion with `hop >= maxHop`, and each relay increments the +// value, so a negative start would delay that cutoff (a huge negative one +// effectively forever), weakening the loop-safety bound exactly when +// membership skew creates a relay cycle. +func parseHop(h string) (int, bool) { + if h == "" { + return 0, true + } + hop, err := strconv.Atoi(h) + if err != nil || hop < 0 { + return 0, false + } + return hop, true +} + // parseBlockPath parses "/peer/v1/block//". func parseBlockPath(p string) (store.BlockKey, bool) { rest, ok := strings.CutPrefix(p, blockPath) diff --git a/internal/peer/stream.go b/internal/peer/stream.go index db3d123..a162054 100644 --- a/internal/peer/stream.go +++ b/internal/peer/stream.go @@ -92,7 +92,11 @@ func (s *StreamServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.Error(w, "bad block path", http.StatusBadRequest) return } - hop, _ := strconv.Atoi(r.Header.Get(HeaderHop)) + hop, ok := parseHop(r.Header.Get(HeaderHop)) + if !ok { + http.Error(w, "bad "+HeaderHop+" header", http.StatusBadRequest) + return + } req := BlockRequest{Key: key, URL: r.Header.Get(HeaderOrigin), Hop: hop} // Buffer the head of the stream so a "not held" or an immediate error can