From bb67241430992bbfb1aae8c4989d8040ac26a564 Mon Sep 17 00:00:00 2001 From: akup Date: Tue, 14 Jul 2026 21:33:45 +0300 Subject: [PATCH 1/5] fix(devshard): stop observability from binding protocol version Public GET diffs/stats no longer CreateSession; only owner-signed chat binds. Join proxy rewrites legacy versioned obs URLs internally; versiond serves versionless paths via PG sessions.version (fan-out fallback). --- .../gonka-devshard-observability.json | 2 +- .../cmd/devshardd/session/bind_obs_test.go | 184 ++++++++++++++++ devshard/cmd/devshardd/session/manager.go | 69 +++++- devshard/cmd/devshardd/session/stats.go | 6 +- .../docs/high-availability-architecture.md | 21 +- devshard/docs/inference-lifecycle.md | 5 +- devshard/docs/upgrade.md | 19 +- devshard/paths.go | 39 ++++ devshard/paths_test.go | 24 +++ devshard/server/routes.go | 57 ++++- devshard/server/routes_test.go | 14 ++ devshard/transport/server.go | 88 +++++--- proxy/README.md | 47 ++++ proxy/entrypoint.sh | 22 +- proxy/nginx.unified.conf.template | 4 +- versioned/cmd/versiond/main.go | 16 +- versioned/go.mod | 6 + versioned/go.sum | 26 +++ versioned/internal/proxy/proxy.go | 201 ++++++++++++++++-- versioned/internal/proxy/proxy_test.go | 167 +++++++++++++++ versioned/internal/sessionversion/lookup.go | 100 +++++++++ 21 files changed, 1033 insertions(+), 84 deletions(-) create mode 100644 devshard/cmd/devshardd/session/bind_obs_test.go create mode 100644 versioned/internal/sessionversion/lookup.go diff --git a/deploy/join/observability/grafana/dashboards/gonka-devshard-observability.json b/deploy/join/observability/grafana/dashboards/gonka-devshard-observability.json index 3835e34e57..07e1b3b615 100644 --- a/deploy/join/observability/grafana/dashboards/gonka-devshard-observability.json +++ b/deploy/join/observability/grafana/dashboards/gonka-devshard-observability.json @@ -9,7 +9,7 @@ }, "id": 1, "options": { - "content": "Use this dashboard to answer one question quickly: where did a devshard inference stop? Start with Terminal Outcomes and Operator Interruptions, then use logs by request_id for the exact request and failure_where.", + "content": "Use this dashboard to answer one question quickly: where did a devshard inference stop? Start with Terminal Outcomes and Operator Interruptions, then use logs by request_id for the exact request and failure_where.\n\nHTTP shard debugging (not this dashboard): prefer versionless `GET /devshard/sessions/{id}/diffs` and `GET /devshard/stats/shards/{id}` — see versionless-observability-plan.md. Legacy `/devshard/{version}/…` obs URLs still work via join-proxy rewrite.", "mode": "markdown" }, "title": "How to Read This", diff --git a/devshard/cmd/devshardd/session/bind_obs_test.go b/devshard/cmd/devshardd/session/bind_obs_test.go new file mode 100644 index 0000000000..a1b4780759 --- /dev/null +++ b/devshard/cmd/devshardd/session/bind_obs_test.go @@ -0,0 +1,184 @@ +package session + +import ( + "bytes" + "encoding/hex" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/require" + + "devshard/bridge" + "devshard/internal/testutil" + "devshard/signing" + "devshard/storage" + "devshard/stub" + "devshard/transport" +) + +func setupBindTestManager(t *testing.T, escrowID string) (*HostManager, *storage.SQLite, *signing.Secp256k1Signer, *signing.Secp256k1Signer) { + t.Helper() + store := newManagerTestStore(t) + hosts := make([]*signing.Secp256k1Signer, 3) + for i := range hosts { + hosts[i] = mustGenerateKey(t) + } + user := mustGenerateKey(t) + addresses := make([]string, len(hosts)) + for i, h := range hosts { + addresses[i] = h.Address() + } + br := &mockBridge{ + escrow: &bridge.EscrowInfo{ + EscrowID: escrowID, + EpochID: 7, + Amount: 100000, + CreatorAddress: user.Address(), + Slots: addresses, + TokenPrice: 1, + }, + } + mgr := NewHostManager(store, hosts[0], stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, br, nil, nil) + return mgr, store, user, hosts[0] +} + +func signedPOST(t *testing.T, e *echo.Echo, signer *signing.Secp256k1Signer, path, escrowID string, body []byte) *httptest.ResponseRecorder { + t.Helper() + ts := time.Now().Unix() + sig, err := transport.SignRequest(signer, escrowID, body, ts) + require.NoError(t, err) + req := httptest.NewRequest(http.MethodPost, path, bytes.NewReader(body)) + req.Header.Set(echo.HeaderContentType, "application/json") + req.Header.Set(transport.HeaderSignature, hex.EncodeToString(sig)) + req.Header.Set(transport.HeaderTimestamp, strconv.FormatInt(ts, 10)) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + return rec +} + +func TestObsGET_DoesNotBindSession(t *testing.T) { + const escrowID = "obs-unbound" + mgr, store, _, _ := setupBindTestManager(t, escrowID) + + e := echo.New() + mgr.Register(e.Group("")) + + req := httptest.NewRequest(http.MethodGet, "/sessions/"+escrowID+"/diffs", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + require.Equal(t, http.StatusNotFound, rec.Code, "body: %s", rec.Body.String()) + + _, err := store.GetSessionMeta(escrowID) + require.ErrorIs(t, err, storage.ErrSessionNotFound) + + active, err := store.ListActiveSessions() + require.NoError(t, err) + require.Empty(t, active) +} + +func TestObsMempoolSignatures_DoNotBindSession(t *testing.T) { + const escrowID = "obs-unbound-2" + mgr, store, _, _ := setupBindTestManager(t, escrowID) + e := echo.New() + mgr.Register(e.Group("")) + + for _, path := range []string{ + "/sessions/" + escrowID + "/mempool", + "/sessions/" + escrowID + "/signatures", + } { + req := httptest.NewRequest(http.MethodGet, path, nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + require.Equal(t, http.StatusNotFound, rec.Code, "path=%s body=%s", path, rec.Body.String()) + } + + _, err := store.GetSessionMeta(escrowID) + require.ErrorIs(t, err, storage.ErrSessionNotFound) +} + +func TestGossipUnbound_DoesNotBindSession(t *testing.T) { + const escrowID = "gossip-unbound" + mgr, store, _, hostSigner := setupBindTestManager(t, escrowID) + e := echo.New() + mgr.Register(e.Group("")) + + body := []byte(`{"nonce":1}`) + rec := signedPOST(t, e, hostSigner, "/sessions/"+escrowID+"/gossip/nonce", escrowID, body) + require.Equal(t, http.StatusNotFound, rec.Code, "body: %s", rec.Body.String()) + + _, err := store.GetSessionMeta(escrowID) + require.ErrorIs(t, err, storage.ErrSessionNotFound) +} + +func TestOwnerChat_BindsSession(t *testing.T) { + const escrowID = "owner-bind" + mgr, store, user, _ := setupBindTestManager(t, escrowID) + e := echo.New() + mgr.Register(e.Group("")) + + body := []byte(`{"model":"m","messages":[{"role":"user","content":"hi"}]}`) + rec := signedPOST(t, e, user, "/sessions/"+escrowID+"/chat/completions", escrowID, body) + // Inference may fail downstream; bind must have happened regardless. + meta, err := store.GetSessionMeta(escrowID) + require.NoError(t, err, "owner chat must CreateSession; http=%d body=%s", rec.Code, rec.Body.String()) + require.Equal(t, testutil.RuntimeTestVersion, meta.Version) + require.Equal(t, user.Address(), meta.CreatorAddr) +} + +func TestNonOwnerChat_DoesNotBindSession(t *testing.T) { + const escrowID = "non-owner-bind" + mgr, store, _, hostSigner := setupBindTestManager(t, escrowID) + e := echo.New() + mgr.Register(e.Group("")) + + body := []byte(`{"model":"m","messages":[{"role":"user","content":"hi"}]}`) + rec := signedPOST(t, e, hostSigner, "/sessions/"+escrowID+"/chat/completions", escrowID, body) + require.Equal(t, http.StatusForbidden, rec.Code, "body: %s", rec.Body.String()) + + _, err := store.GetSessionMeta(escrowID) + require.ErrorIs(t, err, storage.ErrSessionNotFound) +} + +func TestSessionServerExisting_NoCreate(t *testing.T) { + const escrowID = "existing-miss" + mgr, store, _, _ := setupBindTestManager(t, escrowID) + + _, err := mgr.SessionServerExisting(escrowID) + require.ErrorIs(t, err, storage.ErrSessionNotFound) + + _, err = store.GetSessionMeta(escrowID) + require.ErrorIs(t, err, storage.ErrSessionNotFound) +} + +func TestGetOrCreate_RecoversBeforeCreate(t *testing.T) { + store := newManagerTestStore(t) + _, user, hostSigner := populateStore(t, store, 2) + + addresses := []string{hostSigner.Address()} + // populateStore uses 3 hosts; rebuild addresses from meta. + meta, err := store.GetSessionMeta("1") + require.NoError(t, err) + addresses = make([]string, len(meta.Group)) + for i, s := range meta.Group { + addresses[i] = s.ValidatorAddress + } + + br := &mockBridge{ + escrow: &bridge.EscrowInfo{ + EscrowID: "1", + EpochID: 7, + Amount: 100000, + CreatorAddress: user.Address(), + Slots: addresses, + }, + } + mgr := NewHostManager(store, hostSigner, stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, br, nil, nil) + + srv, err := mgr.getOrCreate("1") + require.NoError(t, err) + require.Equal(t, uint64(2), srv.Host().SnapshotState().LatestNonce) +} diff --git a/devshard/cmd/devshardd/session/manager.go b/devshard/cmd/devshardd/session/manager.go index b73c45c8aa..dcb96aa0a0 100644 --- a/devshard/cmd/devshardd/session/manager.go +++ b/devshard/cmd/devshardd/session/manager.go @@ -128,10 +128,66 @@ func (m *HostManager) Close() error { } // SessionServer resolves or creates the per-escrow transport server. +// Prefer BindOwnerChat for HTTP chat; this remains for tests and explicit bind. func (m *HostManager) SessionServer(escrowID string) (*transport.Server, error) { return m.getOrCreate(escrowID) } +// SessionServerExisting returns a live or recovered session server. +// It never CreateSession / binds a protocol version. Missing sessions return +// storage.ErrSessionNotFound. +func (m *HostManager) SessionServerExisting(escrowID string) (*transport.Server, error) { + if srv, ok := m.existingServer(escrowID); ok { + return srv, nil + } + srv, err := m.recoverAndStoreSession(escrowID) + if err != nil { + return nil, err + } + return srv, nil +} + +// BindOwnerChat verifies the request as the escrow owner, then returns an +// existing session or binds a new one with this process's boundVersion. +// Auth context (sender + body) is injected for HandleInference. +func (m *HostManager) BindOwnerChat(c echo.Context) (*transport.Server, error) { + escrowID := c.Param("id") + addr, body, err := transport.VerifyPOSTAuth(c, m.verifier, escrowID, 0) + if err != nil { + return nil, err + } + + srv, err := m.SessionServerExisting(escrowID) + if err == nil { + if !srv.IsOwner(addr) { + return nil, echo.NewHTTPError(http.StatusForbidden, "restricted to escrow owner") + } + transport.InjectAuthContext(c, addr, body) + return srv, nil + } + if !errors.Is(err, storage.ErrSessionNotFound) { + return nil, err + } + + escrow, err := m.bridge.GetEscrow(escrowID) + if err != nil { + return nil, fmt.Errorf("get escrow: %w", err) + } + if escrow == nil || escrow.CreatorAddress == "" || addr != escrow.CreatorAddress { + return nil, echo.NewHTTPError(http.StatusForbidden, "restricted to escrow owner") + } + + srv, err = m.getOrCreate(escrowID) + if err != nil { + return nil, err + } + if !srv.IsOwner(addr) { + return nil, echo.NewHTTPError(http.StatusForbidden, "restricted to escrow owner") + } + transport.InjectAuthContext(c, addr, body) + return srv, nil +} + // HandleSettlementFinalized marks the session inactive and drops the live // transport server so RecoverSessions will not resurrect settled escrows. func (m *HostManager) HandleSettlementFinalized(escrowID string) error { @@ -169,7 +225,16 @@ func (m *HostManager) getOrCreate(escrowID string) (*transport.Server, error) { return nil, err } - srv, err := m.create(escrowID) + // Prefer recovering an already-bound session over CreateSession. + srv, err := m.recoverStoredSession(escrowID) + if err == nil { + return m.storeSessionIfAbsent(escrowID, srv), nil + } + if !errors.Is(err, storage.ErrSessionNotFound) { + return nil, err + } + + srv, err = m.create(escrowID) if err != nil { return nil, err } @@ -440,7 +505,7 @@ func (m *HostManager) recoverStoredSession(escrowID string) (*transport.Server, func (m *HostManager) Register(g *echo.Group) { g.GET("/stats/shards", m.handleStatsShards) g.GET("/stats/shards/:escrow_id", m.handleStatsShard) - devshardserver.RegisterLazySessionRoutes(g, m, m) + devshardserver.RegisterLazySessionRoutes(g, m, m, m) } // HandlePayloads serves payloads to validators for devshard validation. diff --git a/devshard/cmd/devshardd/session/stats.go b/devshard/cmd/devshardd/session/stats.go index 3801433e08..ec00a406a2 100644 --- a/devshard/cmd/devshardd/session/stats.go +++ b/devshard/cmd/devshardd/session/stats.go @@ -178,11 +178,7 @@ func (m *HostManager) statsShardDetail(escrowID string, now time.Time) (*statsSh if err != nil { return nil, err } - // Prefer recovering an already-persisted session over create-via-bridge. - if err := m.TryLoadFromStorage(escrowID); err != nil { - return nil, err - } - srv, err := m.SessionServer(escrowID) + srv, err := m.SessionServerExisting(escrowID) if err != nil { return nil, err } diff --git a/devshard/docs/high-availability-architecture.md b/devshard/docs/high-availability-architecture.md index e2012252a9..6433fe857c 100644 --- a/devshard/docs/high-availability-architecture.md +++ b/devshard/docs/high-availability-architecture.md @@ -44,7 +44,9 @@ out to three independently-deployable backends: |---------------|---------|---------| | 22 Tier A `/v1/*` query routes | `edge-api` (or `edge-api-router`) | Read-only chain queries | | Other `/v1/*`, `/api/v1/*` | `dapi` (`api:9000`) | Chat/inference, PoC, payloads, bridge, identity | -| `/devshard//sessions/...` | `versiond` (or `versiond-router`) → `devshardd` | Devshard session protocol | +| `/devshard//sessions/...` (protocol) | `versiond` (or `versiond-router`) → `devshardd` | Chat, gossip, payloads — version binds on owner chat | +| `/devshard/sessions/...`, `/devshard/stats/...`, `/devshard/metrics` | `versiond` → bound/`primary` child | Versionless public observability (no bind) | +| `/devshard//sessions/.../diffs\|mempool\|signatures` (legacy) | join proxy **internal rewrite** → versionless | Backward-compat for scrapers | | `/v1/devshard/*` (legacy) | rewritten → `/devshard/v1/*` → versiond | Backward-compat | | `/chain-rpc`, `/chain-api`, `/chain-grpc` | `chain-node` | Direct chain access | @@ -108,7 +110,14 @@ A supervisor + version-prefix reverse proxy: (`//...`), backed by an `atomic.Value` route table of `version → localhost:port` for **running** children only (`internal/proxy/proxy.go`, `rebuildRoutes`). -- **HTTP:** `:8080`, `GET /healthz` (per-child status) + version-prefix proxy. +- **Versionless observability:** also serves `/sessions/…/diffs|mempool|signatures`, + `/stats/…`, `/metrics` without a version prefix. With shared Postgres + (`PGHOST` / `DATABASE_URL`), session-scoped routes look up + `sessions.version` and forward to that child; unbound → 404. Without PG + (SQLite-only), fan-out across children. See + [versionless-observability-plan.md](./versionless-observability-plan.md). +- **HTTP:** `:8080`, `GET /healthz` (per-child status) + version-prefix / + versionless obs proxy. - **Overrides:** `VERSIOND_OVERRIDE_` (local binary), `VERSIOND_FORCE` (force-run a version). @@ -118,9 +127,13 @@ The standalone devshard **host** process (a versiond child, never a direct compose service). It runs the per-escrow session protocol: - **Routes:** `GET /healthz`, `GET /metrics`, and session routes - `POST /sessions/:id/chat/completions`, `verify-timeout`, `challenge-receipt`, - `gossip/*`, `GET /sessions/:id/{diffs,mempool,signatures,payloads}` + `POST /sessions/:id/chat/completions` (**owner bind**), `verify-timeout`, + `challenge-receipt`, `gossip/*`, + `GET /sessions/:id/{diffs,mempool,signatures}` (observability — never binds), + `GET /sessions/:id/payloads` (validator protocol) (`devshard/cmd/devshardd/server.go`, `devshard/server/routes.go`). + Public observability is also reachable versionless via the join proxy / + versiond (see [versionless-observability-plan.md](./versionless-observability-plan.md)). - **Chain:** gRPC client (`common/chain`) + CometBFT WebSocket for `NewBlock`, `devshard_escrow_created`, `devshard_escrow_settled`; tracks a `chain.Phase` (epoch/height). Bridge queries + dispute submission via diff --git a/devshard/docs/inference-lifecycle.md b/devshard/docs/inference-lifecycle.md index 2d5146b151..fb3c334129 100644 --- a/devshard/docs/inference-lifecycle.md +++ b/devshard/docs/inference-lifecycle.md @@ -62,8 +62,9 @@ zero, keeping the wire format bit-identical with returning `RequiredValidations` / `CompletedValidations` constantly at zero. - Host-local **validation observability** (outside the state root, exposed - on `GET /v1/devshard/stats/shards/{escrow_id}`) is populated when signed - diffs are applied; see + on `GET /devshard/stats/shards/{escrow_id}` — prefer the versionless path; + legacy `/devshard/{version}/stats/shards/...` is rewritten internally) is + populated when signed diffs are applied; see [validation-observability-diff-apply.md](./validation-observability-diff-apply.md). ## 2. Drop inference records from RAM and state diff --git a/devshard/docs/upgrade.md b/devshard/docs/upgrade.md index af8bc31953..4c977f8f7f 100644 --- a/devshard/docs/upgrade.md +++ b/devshard/docs/upgrade.md @@ -65,19 +65,26 @@ tested or forced locally. Escrow creation stays version-agnostic. `MsgCreateDevshardEscrow` does not take a version. -The user chooses a version by selecting the HTTP path at session start: +The **escrow creator** chooses a version by calling a versioned protocol path +with a valid owner signature. Bind is the first successful owner +`POST /devshard//sessions/{id}/chat/completions` (see +[versionless-observability-plan.md](./versionless-observability-plan.md)). ``` -/devshard//* -> versiond -> devshard binary for +/devshard//sessions/.../chat/completions → versiond → bind + protocol +/devshard/sessions/.../diffs|mempool|signatures → versionless observability (no bind) ``` -The target safety model is that the first request binds the session to one -binary version off-chain. Every later diff must continue with that same +Legacy versioned observability URLs are rewritten internally by the join proxy +to the versionless forms; the version segment in those URLs does not bind. + +The target safety model is that the first **owner** request binds the session to +one binary version off-chain. Every later diff must continue with that same version. A host running the wrong binary refuses to sign, so a version-mixing session cannot gather the threshold needed to settle. -The bound version is recorded in shard state. Use the `` segment from -`/devshard//*`. +The bound version is recorded in shard state / storage (`sessions.version`). +Use the `` segment from `/devshard//*` for protocol traffic only. ## Deprecation diff --git a/devshard/paths.go b/devshard/paths.go index a75801a631..d185b42545 100644 --- a/devshard/paths.go +++ b/devshard/paths.go @@ -68,3 +68,42 @@ func SessionPayloadPath(routePrefix, escrowID string) string { func VersionedSessionPayloadPath(version, escrowID string) string { return SessionPayloadPath(VersionedRoutePrefix(version), escrowID) } + +// VersionlessRoutePrefix is the canonical mount for public observability +// (no protocol version segment). Protocol traffic stays under VersionedRoutePrefix. +const VersionlessRoutePrefix = "/devshard" + +// VersionlessSessionDiffsPath is GET /devshard/sessions/{id}/diffs. +func VersionlessSessionDiffsPath(escrowID string) string { + return fmt.Sprintf("%s/sessions/%s/diffs", VersionlessRoutePrefix, escrowID) +} + +// VersionlessSessionMempoolPath is GET /devshard/sessions/{id}/mempool. +func VersionlessSessionMempoolPath(escrowID string) string { + return fmt.Sprintf("%s/sessions/%s/mempool", VersionlessRoutePrefix, escrowID) +} + +// VersionlessSessionSignaturesPath is GET /devshard/sessions/{id}/signatures. +func VersionlessSessionSignaturesPath(escrowID string) string { + return fmt.Sprintf("%s/sessions/%s/signatures", VersionlessRoutePrefix, escrowID) +} + +// VersionlessStatsShardsPath is GET /devshard/stats/shards. +func VersionlessStatsShardsPath() string { + return VersionlessRoutePrefix + "/stats/shards" +} + +// VersionlessStatsShardDetailPath is GET /devshard/stats/shards/{id}. +func VersionlessStatsShardDetailPath(escrowID string) string { + return fmt.Sprintf("%s/stats/shards/%s", VersionlessRoutePrefix, escrowID) +} + +// VersionlessMetricsPath is GET /devshard/metrics. +func VersionlessMetricsPath() string { + return VersionlessRoutePrefix + "/metrics" +} + +// VersionlessHealthzPath is GET /devshard/healthz. +func VersionlessHealthzPath() string { + return VersionlessRoutePrefix + "/healthz" +} diff --git a/devshard/paths_test.go b/devshard/paths_test.go index 7a1510d204..a4b7497cb5 100644 --- a/devshard/paths_test.go +++ b/devshard/paths_test.go @@ -167,3 +167,27 @@ func TestSessionPayloadPath(t *testing.T) { }) } } + +func TestVersionlessObservabilityPaths(t *testing.T) { + if got := VersionlessSessionDiffsPath("42"); got != "/devshard/sessions/42/diffs" { + t.Fatalf("diffs = %q", got) + } + if got := VersionlessSessionMempoolPath("42"); got != "/devshard/sessions/42/mempool" { + t.Fatalf("mempool = %q", got) + } + if got := VersionlessSessionSignaturesPath("42"); got != "/devshard/sessions/42/signatures" { + t.Fatalf("signatures = %q", got) + } + if got := VersionlessStatsShardsPath(); got != "/devshard/stats/shards" { + t.Fatalf("stats = %q", got) + } + if got := VersionlessStatsShardDetailPath("42"); got != "/devshard/stats/shards/42" { + t.Fatalf("stats detail = %q", got) + } + if got := VersionlessMetricsPath(); got != "/devshard/metrics" { + t.Fatalf("metrics = %q", got) + } + if got := VersionlessHealthzPath(); got != "/devshard/healthz" { + t.Fatalf("healthz = %q", got) + } +} diff --git a/devshard/server/routes.go b/devshard/server/routes.go index a0d4949371..58a5bda3a5 100644 --- a/devshard/server/routes.go +++ b/devshard/server/routes.go @@ -15,9 +15,15 @@ import ( // ErrInitializing means devshard storage is not ready to serve session state yet. var ErrInitializing = errors.New("devshard initializing") -// SessionResolver resolves a lazy per-escrow transport server. +// SessionResolver resolves an already-bound per-escrow transport server. +// Implementations must never CreateSession / bind a protocol version. type SessionResolver interface { - SessionServer(escrowID string) (*transport.Server, error) + SessionServerExisting(escrowID string) (*transport.Server, error) +} + +// OwnerChatBinder authenticates the escrow owner and may bind a new session. +type OwnerChatBinder interface { + BindOwnerChat(c echo.Context) (*transport.Server, error) } // PayloadHandler serves GET /sessions/:id/payloads for a resolved session. @@ -26,12 +32,13 @@ type PayloadHandler interface { } // RegisterLazySessionRoutes mounts the standard devshard HTTP surface on g. -// Session servers are resolved lazily per request via SessionResolver. -func RegisterLazySessionRoutes(g *echo.Group, resolver SessionResolver, payloadHandler PayloadHandler) { +// Observability and host protocol routes resolve existing sessions only. +// Only owner chat may bind a new session (via OwnerChatBinder). +func RegisterLazySessionRoutes(g *echo.Group, resolver SessionResolver, binder OwnerChatBinder, payloadHandler PayloadHandler) { g.Use(observability.EchoMiddleware()) g.Use(observability.RequestIDMiddleware) - g.POST("/sessions/:id/chat/completions", withSessionAuth(resolver, true, + g.POST("/sessions/:id/chat/completions", withOwnerChat(binder, true, func(srv *transport.Server) echo.HandlerFunc { return srv.HandleInference })) g.POST("/sessions/:id/verify-timeout", withSessionAuth(resolver, false, func(srv *transport.Server) echo.HandlerFunc { return srv.HandleVerifyTimeout })) @@ -51,7 +58,7 @@ func RegisterLazySessionRoutes(g *echo.Group, resolver SessionResolver, payloadH if payloadHandler != nil { g.GET("/sessions/:id/payloads", func(c echo.Context) error { - srv, err := resolver.SessionServer(c.Param("id")) + srv, err := resolver.SessionServerExisting(c.Param("id")) if err != nil { recordSessionResolution(c, err, false) return sessionHTTPError(c, err) @@ -67,7 +74,7 @@ func withSession( pick func(*transport.Server) echo.HandlerFunc, ) echo.HandlerFunc { return func(c echo.Context) error { - srv, err := resolver.SessionServer(c.Param("id")) + srv, err := resolver.SessionServerExisting(c.Param("id")) if err != nil { recordSessionResolution(c, err, false) return sessionHTTPError(c, err) @@ -83,7 +90,7 @@ func withSessionAuth( pick func(*transport.Server) echo.HandlerFunc, ) echo.HandlerFunc { return func(c echo.Context) error { - srv, err := resolver.SessionServer(c.Param("id")) + srv, err := resolver.SessionServerExisting(c.Param("id")) if err != nil { recordSessionResolution(c, err, recordChatTerminal) return sessionHTTPError(c, err) @@ -95,6 +102,23 @@ func withSessionAuth( } } +func withOwnerChat( + binder OwnerChatBinder, + recordChatTerminal bool, + pick func(*transport.Server) echo.HandlerFunc, +) echo.HandlerFunc { + return func(c echo.Context) error { + srv, err := binder.BindOwnerChat(c) + if err != nil { + recordSessionResolution(c, err, recordChatTerminal) + return sessionHTTPError(c, err) + } + observability.IncSessionResolution(routeLabel(c), observability.MetricStatusOK, observability.ReasonOK) + handler := pick(srv) + return srv.RateLimitMiddleware(recordChatTerminal)(handler)(c) + } +} + func recordSessionResolution(c echo.Context, err error, recordChatTerminal bool) { metricStatus, reason := sessionResolutionStatus(err) route := routeLabel(c) @@ -111,12 +135,22 @@ func sessionResolutionStatus(err error) (observability.MetricStatus, observabili if errors.Is(err, ErrInitializing) { return observability.MetricStatusError, observability.ReasonInitializing } + if errors.Is(err, storage.ErrSessionNotFound) { + return observability.MetricStatusError, observability.ReasonSessionResolveErr + } if errors.Is(err, storage.ErrSessionVersionConflict) { return observability.MetricStatusError, observability.ReasonVersionConflict } if errors.Is(err, storage.ErrSessionEpochConflict) { return observability.MetricStatusError, observability.ReasonEpochConflict } + var he *echo.HTTPError + if errors.As(err, &he) { + switch he.Code { + case http.StatusUnauthorized, http.StatusForbidden: + return observability.MetricStatusError, observability.ReasonOwnerErr + } + } msg := err.Error() switch { case strings.Contains(msg, "build group"): @@ -152,9 +186,16 @@ func routeLabel(c echo.Context) string { } func sessionHTTPError(c echo.Context, err error) error { + var he *echo.HTTPError + if errors.As(err, &he) { + return he + } if errors.Is(err, ErrInitializing) { return transport.HTTPError(c, http.StatusServiceUnavailable, transport.DevshardErrorInitializing, err.Error()) } + if errors.Is(err, storage.ErrSessionNotFound) { + return echo.NewHTTPError(http.StatusNotFound, "session not found") + } if errors.Is(err, storage.ErrSessionVersionConflict) || errors.Is(err, storage.ErrSessionEpochConflict) { return echo.NewHTTPError(http.StatusConflict, err.Error()) } diff --git a/devshard/server/routes_test.go b/devshard/server/routes_test.go index 52d3cb42b2..80ac4120ca 100644 --- a/devshard/server/routes_test.go +++ b/devshard/server/routes_test.go @@ -47,6 +47,20 @@ func TestSessionHTTPErrorInitializing(t *testing.T) { require.Equal(t, transport.DevshardErrorInitializing, rec.Header().Get(transport.HeaderDevshardError)) } +func TestSessionHTTPErrorNotFound(t *testing.T) { + c := testEchoContext(t) + httpErr, ok := sessionHTTPError(c, storage.ErrSessionNotFound).(*echo.HTTPError) + require.True(t, ok) + require.Equal(t, http.StatusNotFound, httpErr.Code) +} + +func TestSessionHTTPErrorPassthroughHTTPError(t *testing.T) { + c := testEchoContext(t) + orig := echo.NewHTTPError(http.StatusForbidden, "restricted to escrow owner") + got := sessionHTTPError(c, orig) + require.Equal(t, orig, got) +} + func TestSessionHTTPErrorDefault(t *testing.T) { c := testEchoContext(t) httpErr, ok := sessionHTTPError(c, fmt.Errorf("boom")).(*echo.HTTPError) diff --git a/devshard/transport/server.go b/devshard/transport/server.go index ad165953f8..ad5574ad21 100644 --- a/devshard/transport/server.go +++ b/devshard/transport/server.go @@ -174,6 +174,54 @@ func (s *Server) isOwner(addr string) bool { return s.userAddr != "" && addr == s.userAddr } +// IsOwner reports whether addr is the escrow creator for this session. +func (s *Server) IsOwner(addr string) bool { + return s.isOwner(addr) +} + +// InjectAuthContext stores a verified sender and request body for handlers +// that run after auth (HandleInference, gossip, etc.). +func InjectAuthContext(c echo.Context, sender string, body []byte) { + c.Set(contextKeySender, sender) + c.Set("body", body) +} + +// VerifyPOSTAuth reads signature headers and body, verifies the signature, and +// returns the recovered sender address and body. It does not check group +// membership or ownership — callers enforce that. +func VerifyPOSTAuth(c echo.Context, verifier signing.Verifier, escrowID string, maxBodySize int64) (string, []byte, error) { + sigHex := c.Request().Header.Get(HeaderSignature) + tsStr := c.Request().Header.Get(HeaderTimestamp) + if sigHex == "" || tsStr == "" { + return "", nil, echo.NewHTTPError(http.StatusUnauthorized, "missing auth headers") + } + + sig, err := hex.DecodeString(sigHex) + if err != nil { + return "", nil, echo.NewHTTPError(http.StatusUnauthorized, "invalid signature hex") + } + + ts, err := strconv.ParseInt(tsStr, 10, 64) + if err != nil { + return "", nil, echo.NewHTTPError(http.StatusUnauthorized, "invalid timestamp") + } + + if maxBodySize > 0 { + c.Request().Body = http.MaxBytesReader(c.Response(), c.Request().Body, maxBodySize) + } + + body, err := io.ReadAll(c.Request().Body) + if err != nil { + return "", nil, echo.NewHTTPError(http.StatusBadRequest, "read body") + } + + addr, err := VerifyRequest(verifier, escrowID, body, sig, ts, time.Now().Unix()) + if err != nil { + return "", nil, echo.NewHTTPError(http.StatusUnauthorized, err.Error()) + } + return addr, body, nil +} + // isGroupMember returns true if addr is a group member or a warm key for // a group member (excludes the user). Gossip is host-to-host; the user has // no business gossiping. @@ -184,55 +232,25 @@ func (s *Server) isGroupMember(addr string) bool { return s.isWarmKeySender(addr) } -// authMiddleware reads the body, verifies the signature, checks group membership, +// AuthMiddleware reads the body, verifies the signature, checks group membership, // and stores the sender address in the echo context. -// GET requests skip auth intentionally for now. +// GET requests skip auth intentionally (public observability). func (s *Server) AuthMiddleware(next echo.HandlerFunc) echo.HandlerFunc { return func(c echo.Context) error { if c.Request().Method == http.MethodGet { - // GET endpoints skip auth for now -- see Register comment. return next(c) } - sigHex := c.Request().Header.Get(HeaderSignature) - tsStr := c.Request().Header.Get(HeaderTimestamp) - if sigHex == "" || tsStr == "" { - return echo.NewHTTPError(http.StatusUnauthorized, "missing auth headers") - } - - sig, err := hex.DecodeString(sigHex) - if err != nil { - return echo.NewHTTPError(http.StatusUnauthorized, "invalid signature hex") - } - - ts, err := strconv.ParseInt(tsStr, 10, 64) - if err != nil { - return echo.NewHTTPError(http.StatusUnauthorized, "invalid timestamp") - } - - // Cap body size before reading. - if s.maxBodySize > 0 { - c.Request().Body = http.MaxBytesReader(c.Response(), c.Request().Body, s.maxBodySize) - } - - body, err := io.ReadAll(c.Request().Body) - if err != nil { - return echo.NewHTTPError(http.StatusBadRequest, "read body") - } - - now := time.Now().Unix() - addr, err := VerifyRequest(s.verifier, s.host.EscrowID(), body, sig, ts, now) + addr, body, err := VerifyPOSTAuth(c, s.verifier, s.host.EscrowID(), s.maxBodySize) if err != nil { - return echo.NewHTTPError(http.StatusUnauthorized, err.Error()) + return err } if !s.isAllowedSender(addr) { return echo.NewHTTPError(http.StatusForbidden, "sender not in group") } - // Store sender and re-inject body for handler. - c.Set(contextKeySender, addr) - c.Set("body", body) + InjectAuthContext(c, addr, body) return next(c) } } diff --git a/proxy/README.md b/proxy/README.md index 448d3c3fb6..728a82a091 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -115,6 +115,53 @@ Key runtime environment variables: | `VERSIOND_PORT` | 8080 | Port on the versiond (or versiond-router) upstream. | | `DISABLE_DEVSHARD_PROXY` | false | Set to `true` to disable `/devshard/` and `/v1/devshard/` routing to versiond. | +Versiond-side (on the versiond container, not the join proxy): + +| Env | Default | Description | +|-----|---------|-------------| +| `PGHOST` / `DATABASE_URL` | unset | When set, versiond looks up `sessions.version` for versionless session obs. | +| `VERSIOND_DISABLE_SESSION_LOOKUP` | false | Force fan-out even if Postgres is configured. | + +### Devshard observability routing + +Public observability paths that still include a version segment are **rewritten +internally** to versionless canonical URIs (no client-visible redirect). Clients +keep calling the old URLs; nginx drops the version segment before versiond so +scrapers need not follow redirects and cannot bind protocol version via the path. + +**Prefer these URLs in new monitors / runbooks:** + +```text +GET /devshard/sessions/{escrow_id}/diffs +GET /devshard/sessions/{escrow_id}/mempool +GET /devshard/sessions/{escrow_id}/signatures +GET /devshard/stats/shards +GET /devshard/stats/shards/{escrow_id} +GET /devshard/metrics +``` + +| Client URL (legacy, still works) | Internal route to versiond | +|----------------------------------|----------------------------| +| `GET /devshard/{version}/sessions/{id}/diffs` | `/devshard/sessions/{id}/diffs` | +| `GET /devshard/{version}/sessions/{id}/mempool` | `/devshard/sessions/{id}/mempool` | +| `GET /devshard/{version}/sessions/{id}/signatures` | `/devshard/sessions/{id}/signatures` | +| `GET /devshard/{version}/stats/shards…` | `/devshard/stats/shards…` | +| `GET /devshard/{version}/metrics` | `/devshard/metrics` | +| `GET /devshard/{version}/healthz` | `/devshard/healthz` | + +Protocol traffic stays versioned: `POST …/chat/completions`, gossip, challenge-receipt, and `GET …/payloads` are **not** rewritten. + +versiond serves the versionless obs paths: + +- **Session-scoped** (`/sessions/{id}/diffs|mempool|signatures`, `/stats/shards/{id}`): when Postgres is configured (`PGHOST` / `DATABASE_URL`), route by `sessions.version`; unbound → 404. If lookup is disabled or PG errors, fan-out across children. +- **Process-level** (`/metrics`, `/healthz`, `/stats/shards` list): pin to lexicographic-max running version (list is not merged across versions). + +Disable lookup: `VERSIOND_DISABLE_SESSION_LOOKUP=true`. + +Grafana dashboards in `deploy/join/observability/` scrape Prometheus metrics from +devshardd `/metrics` via service discovery — they do not call HTTP diffs URLs. +For ad-hoc HTTP debugging of a shard, use the versionless paths above. + ### edge-api vs dapi routing `/v1/` is split across two backends. The proxy registers **exact/regex locations for read-only query paths** before the generic `/v1/` catch-all: diff --git a/proxy/entrypoint.sh b/proxy/entrypoint.sh index 3af269004a..6f57e31bb3 100755 --- a/proxy/entrypoint.sh +++ b/proxy/entrypoint.sh @@ -602,8 +602,24 @@ fi # /devshard/ location -- forwards to versiond which dispatches to the matching # child binary. Treated as exempt (inference forwarding): streaming, long # timeouts, exempt rate/conn limits, CORS. +# +# Phase 1: versioned observability paths are rewritten internally to versionless +# canonical URLs (no client-visible redirect). Dashboards that hardcode +# /devshard/{version}/sessions/.../diffs keep working; the version segment is +# dropped before versiond so it cannot participate in protocol bind. Protocol +# POSTs and payloads stay on /devshard/{version}/... via the catch-all below. if [ "${DISABLE_DEVSHARD_PROXY}" != "true" ]; then - export DEVSHARD_VERSIOND_LOCATION="location /devshard/ { + export DEVSHARD_VERSIOND_LOCATION="# Versioned obs → versionless (internal rewrite); protocol stays versioned + location ~ ^/devshard/[^/]+/sessions/([^/]+)/(diffs|mempool|signatures)\$ { + rewrite ^ /devshard/sessions/\$\$1/\$\$2 last; + } + location ~ ^/devshard/[^/]+/stats/shards(/.*)?\$ { + rewrite ^ /devshard/stats/shards\$\$1 last; + } + location ~ ^/devshard/[^/]+/(metrics|healthz)\$ { + rewrite ^ /devshard/\$\$1 last; + } + location /devshard/ { set \$limit_zone_name \"EXEMPT\"; limit_req zone=exempt_zone burst=${EXEMPT_BURST} nodelay; ${LIMIT_CONN_RULE_EXEMPT} @@ -1201,7 +1217,9 @@ if [ -n "${EDGE_API_SERVICE_NAME}" ]; then fi if [ "${DISABLE_DEVSHARD_PROXY}" != "true" ]; then echo " /devshard/* -> Versiond (devshard binaries)" - echo " /v1/devshard/* -> /devshard/v1/* (legacy redirect)" + echo " /devshard/{v}/sessions/*/diffs|mempool|signatures -> rewrite /devshard/sessions/..." + echo " /devshard/{v}/stats/* /metrics /healthz -> rewrite versionless (internal)" + echo " /v1/devshard/* -> /devshard/v1/* (legacy rewrite)" fi echo " /health -> Health check" diff --git a/proxy/nginx.unified.conf.template b/proxy/nginx.unified.conf.template index 295b63041e..03e6a19df4 100644 --- a/proxy/nginx.unified.conf.template +++ b/proxy/nginx.unified.conf.template @@ -102,7 +102,9 @@ http { # Versiond upstream - only included if VERSIOND_SERVICE_NAME is set. # Routes /devshard//* to versiond which forwards to the matching - # devshard binary. + # devshard binary. Observability paths under /devshard//… are + # rewritten internally to versionless /devshard/sessions|stats|metrics|healthz + # (see entrypoint); no client-visible redirect. ${VERSIOND_UPSTREAM} # Enable gzip compression diff --git a/versioned/cmd/versiond/main.go b/versioned/cmd/versiond/main.go index 342a65121f..2e3e8b6f1f 100644 --- a/versioned/cmd/versiond/main.go +++ b/versioned/cmd/versiond/main.go @@ -16,6 +16,7 @@ import ( "versioned/internal/oracle" "versioned/internal/process" "versioned/internal/proxy" + "versioned/internal/sessionversion" ) func main() { @@ -42,9 +43,22 @@ func run(ctx context.Context) error { mgr := process.NewManager(cfg) oracleClient := oracle.NewClient(cfg.OracleURL) + lookup, err := sessionversion.OpenFromEnv(ctx) + if err != nil { + slog.Warn("session version lookup unavailable; versionless obs will fan-out", "error", err) + lookup = nil + } + if lookup != nil { + defer lookup.Close() + } + mux := http.NewServeMux() mux.HandleFunc("/healthz", health.Handler(mgr.Status)) - mux.Handle("/", proxy.Handler(mgr.RouteTable())) + var proxyOpts []proxy.HandlerOption + if lookup != nil { + proxyOpts = append(proxyOpts, proxy.WithSessionVersionLookup(lookup)) + } + mux.Handle("/", proxy.Handler(mgr.RouteTable(), proxyOpts...)) listenAddr := config.ListenAddr() srv := &http.Server{ diff --git a/versioned/go.mod b/versioned/go.mod index 6e55cc4938..81e179fc66 100644 --- a/versioned/go.mod +++ b/versioned/go.mod @@ -3,12 +3,18 @@ module versioned go 1.24.2 require ( + github.com/jackc/pgx/v5 v5.7.4 google.golang.org/grpc v1.80.0 google.golang.org/protobuf v1.36.11 ) require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + golang.org/x/crypto v0.47.0 // indirect golang.org/x/net v0.49.0 // indirect + golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.40.0 // indirect golang.org/x/text v0.33.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516 // indirect diff --git a/versioned/go.sum b/versioned/go.sum index 98ff7bd506..138ae0700a 100644 --- a/versioned/go.sum +++ b/versioned/go.sum @@ -1,5 +1,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -10,6 +13,21 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.7.4 h1:9wKznZrhWa2QiHL+NjTSPP6yjl3451BX3imWDnokYlg= +github.com/jackc/pgx/v5 v5.7.4/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= @@ -22,8 +40,12 @@ go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2W go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= +golang.org/x/crypto v0.47.0 h1:V6e3FRj+n4dbpw86FJ8Fv7XVOql7TEwpHapKoMJ/GO8= +golang.org/x/crypto v0.47.0/go.mod h1:ff3Y9VzzKbwSSEzWqJsJVBnWmRwRSHt/6Op5n9bQc4A= golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= @@ -36,3 +58,7 @@ google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/versioned/internal/proxy/proxy.go b/versioned/internal/proxy/proxy.go index ea68131dc6..4de1eea125 100644 --- a/versioned/internal/proxy/proxy.go +++ b/versioned/internal/proxy/proxy.go @@ -1,57 +1,224 @@ package proxy import ( + "context" "fmt" "net/http" + "net/http/httptest" "net/http/httputil" "net/url" + "sort" "strings" "sync/atomic" ) +// SessionVersionLookup resolves which protocol version owns a bound escrow. +// ok=false means the escrow is unbound (no session row). +type SessionVersionLookup interface { + LookupSessionVersion(ctx context.Context, escrowID string) (version string, ok bool, err error) +} + +// HandlerOption configures versionless observability routing. +type HandlerOption func(*handlerConfig) + +type handlerConfig struct { + lookup SessionVersionLookup +} + +// WithSessionVersionLookup enables bound-version routing for session-scoped +// versionless observability. When unset, session obs falls back to fan-out. +func WithSessionVersionLookup(lookup SessionVersionLookup) HandlerOption { + return func(c *handlerConfig) { + c.lookup = lookup + } +} + // Handler returns an http.Handler that routes requests by version prefix. // First path segment is the version name, stripped before forwarding. // Example: /v0.2.11/chat/completions -> localhost:9001/chat/completions -func Handler(routes *atomic.Value) http.Handler { +// +// Versionless observability paths (sessions/.../diffs|mempool|signatures, +// stats/..., metrics, healthz) are forwarded without a version prefix so join +// proxy can rewrite legacy /{version}/obs URLs onto canonical versionless URLs. +func Handler(routes *atomic.Value, opts ...HandlerOption) http.Handler { + var cfg handlerConfig + for _, opt := range opts { + opt(&cfg) + } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { path := strings.TrimPrefix(r.URL.Path, "/") - parts := strings.SplitN(path, "/", 2) - if len(parts) == 0 || parts[0] == "" { + if path == "" { http.Error(w, "version prefix required", http.StatusBadRequest) return } + routeMap := routes.Load().(map[string]string) + + if isVersionlessObsPath(path) { + serveVersionlessObs(w, r, routeMap, "/"+path, cfg.lookup) + return + } + + parts := strings.SplitN(path, "/", 2) version := parts[0] rest := "/" if len(parts) == 2 { rest = "/" + parts[1] } - routeMap := routes.Load().(map[string]string) target, ok := routeMap[version] if !ok { http.Error(w, fmt.Sprintf("version %q not found", version), http.StatusNotFound) return } - targetURL, err := url.Parse("http://" + target) + reverseProxy(target, rest).ServeHTTP(w, r) + }) +} + +// isVersionlessObsPath reports public observability paths that must not require +// a protocol version segment. payloads stays versioned (validator protocol). +func isVersionlessObsPath(path string) bool { + switch path { + case "metrics", "healthz": + return true + } + if path == "stats" || strings.HasPrefix(path, "stats/") { + return true + } + parts := strings.Split(path, "/") + if len(parts) == 3 && parts[0] == "sessions" { + switch parts[2] { + case "diffs", "mempool", "signatures": + return true + } + } + return false +} + +func serveVersionlessObs(w http.ResponseWriter, r *http.Request, routeMap map[string]string, rest string, lookup SessionVersionLookup) { + versions := sortedVersions(routeMap) + if len(versions) == 0 { + http.Error(w, "no versions available", http.StatusServiceUnavailable) + return + } + + // Process-level obs (/metrics, /healthz, /stats/shards list): pin to primary. + // Multi-version aggregation of /stats/shards is deferred; primary is the + // lexicographic-max approved version (documented limitation for non-HA / multi-version lists). + escrowID, scoped := escrowIDFromObsPath(rest) + if !scoped { + reverseProxy(routeMap[versions[len(versions)-1]], rest).ServeHTTP(w, r) + return + } + + if lookup != nil { + ver, ok, err := lookup.LookupSessionVersion(r.Context(), escrowID) if err != nil { - http.Error(w, "internal error", http.StatusInternalServerError) + // Shared index unavailable — degrade to fan-out. + serveSessionObsFanout(w, r, routeMap, versions, rest) + return + } + if !ok || ver == "" { + http.NotFound(w, r) + return + } + target, found := routeMap[ver] + if !found { + http.Error(w, fmt.Sprintf("bound version %q not running", ver), http.StatusNotFound) return } + reverseProxy(target, rest).ServeHTTP(w, r) + return + } - p := &httputil.ReverseProxy{ - Rewrite: func(pr *httputil.ProxyRequest) { - pr.SetXForwarded() - pr.Out.URL.Scheme = targetURL.Scheme - pr.Out.URL.Host = targetURL.Host - pr.Out.Host = targetURL.Host - pr.Out.URL.Path = rest - pr.Out.URL.RawPath = "" + serveSessionObsFanout(w, r, routeMap, versions, rest) +} + +func serveSessionObsFanout(w http.ResponseWriter, r *http.Request, routeMap map[string]string, versions []string, rest string) { + var fallback409 *httptest.ResponseRecorder + for _, ver := range versions { + rec := httptest.NewRecorder() + reverseProxy(routeMap[ver], rest).ServeHTTP(rec, r.Clone(r.Context())) + switch { + case rec.Code == http.StatusNotFound: + continue + case rec.Code == http.StatusConflict: + fallback409 = rec + continue + default: + writeRecorder(w, rec) + return + } + } + if fallback409 != nil { + writeRecorder(w, fallback409) + return + } + http.NotFound(w, r) +} + +// escrowIDFromObsPath extracts the escrow id from session-scoped or stats-detail +// versionless paths. ok=false for process-level paths (metrics, shard list, …). +func escrowIDFromObsPath(rest string) (string, bool) { + path := strings.TrimPrefix(rest, "/") + parts := strings.Split(path, "/") + if len(parts) == 3 && parts[0] == "sessions" { + switch parts[2] { + case "diffs", "mempool", "signatures": + if parts[1] != "" { + return parts[1], true + } + } + } + // /stats/shards/{escrow_id} + if len(parts) == 3 && parts[0] == "stats" && parts[1] == "shards" && parts[2] != "" { + return parts[2], true + } + return "", false +} + +func sortedVersions(routeMap map[string]string) []string { + versions := make([]string, 0, len(routeMap)) + for v := range routeMap { + versions = append(versions, v) + } + sort.Strings(versions) + return versions +} + +func reverseProxy(target, rest string) *httputil.ReverseProxy { + targetURL, err := url.Parse("http://" + target) + if err != nil { + return &httputil.ReverseProxy{ + Director: func(req *http.Request) { + req.URL.Scheme = "http" + req.URL.Host = "invalid" + }, + ErrorHandler: func(w http.ResponseWriter, _ *http.Request, _ error) { + http.Error(w, "internal error", http.StatusInternalServerError) }, - FlushInterval: -1, // flush immediately for SSE } + } + return &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + pr.SetXForwarded() + pr.Out.URL.Scheme = targetURL.Scheme + pr.Out.URL.Host = targetURL.Host + pr.Out.Host = targetURL.Host + pr.Out.URL.Path = rest + pr.Out.URL.RawPath = "" + }, + FlushInterval: -1, // flush immediately for SSE + } +} - p.ServeHTTP(w, r) - }) +func writeRecorder(w http.ResponseWriter, rec *httptest.ResponseRecorder) { + for k, vals := range rec.Header() { + for _, v := range vals { + w.Header().Add(k, v) + } + } + w.WriteHeader(rec.Code) + _, _ = w.Write(rec.Body.Bytes()) } diff --git a/versioned/internal/proxy/proxy_test.go b/versioned/internal/proxy/proxy_test.go index b2e6a3061f..dbae426e73 100644 --- a/versioned/internal/proxy/proxy_test.go +++ b/versioned/internal/proxy/proxy_test.go @@ -2,6 +2,7 @@ package proxy import ( "bufio" + "context" "fmt" "io" "net/http" @@ -103,6 +104,172 @@ func TestProxy_NoVersionPrefix(t *testing.T) { } } +func TestProxy_VersionlessObs_Primary(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "path=%s", r.URL.Path) + })) + defer backend.Close() + + addr := strings.TrimPrefix(backend.URL, "http://") + // Lexicographic max is v2 — process-level obs pins to primary. + routes := newRoutes(map[string]string{"v1": "127.0.0.1:1", "v2": addr}) + handler := Handler(routes) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/metrics") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "path=/metrics" { + t.Fatalf("status=%d body=%q", resp.StatusCode, body) + } +} + +func TestProxy_VersionlessObs_SessionFanout(t *testing.T) { + miss := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.NotFound(w, nil) + })) + defer miss.Close() + hit := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "path=%s", r.URL.Path) + })) + defer hit.Close() + + routes := newRoutes(map[string]string{ + "v1": strings.TrimPrefix(miss.URL, "http://"), + "v2": strings.TrimPrefix(hit.URL, "http://"), + }) + handler := Handler(routes) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/sessions/42/diffs") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "path=/sessions/42/diffs" { + t.Fatalf("status=%d body=%q", resp.StatusCode, body) + } +} + +type mapLookup map[string]string + +func (m mapLookup) LookupSessionVersion(_ context.Context, escrowID string) (string, bool, error) { + v, ok := m[escrowID] + return v, ok, nil +} + +type errLookup struct{ err error } + +func (e errLookup) LookupSessionVersion(context.Context, string) (string, bool, error) { + return "", false, e.err +} + +func TestProxy_VersionlessObs_BoundVersionLookup(t *testing.T) { + v1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "v1:%s", r.URL.Path) + })) + defer v1.Close() + v2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "v2:%s", r.URL.Path) + })) + defer v2.Close() + + routes := newRoutes(map[string]string{ + "v1": strings.TrimPrefix(v1.URL, "http://"), + "v2": strings.TrimPrefix(v2.URL, "http://"), + }) + handler := Handler(routes, WithSessionVersionLookup(mapLookup{"42": "v2"})) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/sessions/42/diffs") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "v2:/sessions/42/diffs" { + t.Fatalf("status=%d body=%q", resp.StatusCode, body) + } + + // stats detail also uses bound-version lookup + resp2, err := http.Get(srv.URL + "/stats/shards/42") + if err != nil { + t.Fatal(err) + } + defer resp2.Body.Close() + body2, _ := io.ReadAll(resp2.Body) + if resp2.StatusCode != http.StatusOK || string(body2) != "v2:/stats/shards/42" { + t.Fatalf("stats detail status=%d body=%q", resp2.StatusCode, body2) + } +} + +func TestProxy_VersionlessObs_LookupUnbound404(t *testing.T) { + hit := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + t.Fatal("backend must not be contacted for unbound escrow") + })) + defer hit.Close() + + routes := newRoutes(map[string]string{"v2": strings.TrimPrefix(hit.URL, "http://")}) + handler := Handler(routes, WithSessionVersionLookup(mapLookup{})) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/sessions/missing/diffs") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("status=%d, want 404", resp.StatusCode) + } +} + +func TestProxy_VersionlessObs_LookupErrorFallsBackToFanout(t *testing.T) { + hit := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "ok:%s", r.URL.Path) + })) + defer hit.Close() + + routes := newRoutes(map[string]string{"v2": strings.TrimPrefix(hit.URL, "http://")}) + handler := Handler(routes, WithSessionVersionLookup(errLookup{err: fmt.Errorf("pg down")})) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/sessions/42/diffs") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "ok:/sessions/42/diffs" { + t.Fatalf("status=%d body=%q", resp.StatusCode, body) + } +} + +func TestProxy_VersionlessObs_PayloadsStillVersioned(t *testing.T) { + routes := newRoutes(map[string]string{"v2": "127.0.0.1:1"}) + handler := Handler(routes) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/sessions/42/payloads") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + // Treated as version name "sessions" → not found in route map. + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("status=%d, want 404 (payloads must stay versioned)", resp.StatusCode) + } +} + func TestProxy_QueryParams(t *testing.T) { backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "query=%s", r.URL.RawQuery) diff --git a/versioned/internal/sessionversion/lookup.go b/versioned/internal/sessionversion/lookup.go new file mode 100644 index 0000000000..7fedb5b9f9 --- /dev/null +++ b/versioned/internal/sessionversion/lookup.go @@ -0,0 +1,100 @@ +package sessionversion + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Lookup resolves escrow → bound protocol version from shared Postgres session +// metadata (devshard_session_index + devshard_sessions). +type Lookup struct { + pool *pgxpool.Pool +} + +// OpenFromEnv connects using libpq env (PGHOST/PGUSER/… or DATABASE_URL). +// Returns (nil, nil) when Postgres is not configured or explicitly disabled. +func OpenFromEnv(ctx context.Context) (*Lookup, error) { + if os.Getenv("VERSIOND_DISABLE_SESSION_LOOKUP") == "true" { + slog.Info("session version lookup disabled by VERSIOND_DISABLE_SESSION_LOOKUP") + return nil, nil + } + if !postgresConfigured() { + slog.Info("session version lookup disabled (no PGHOST/DATABASE_URL); versionless obs uses fan-out") + return nil, nil + } + + connString := os.Getenv("DATABASE_URL") + cfg, err := pgxpool.ParseConfig(connString) + if err != nil { + return nil, fmt.Errorf("parse postgres config: %w", err) + } + cfg.MaxConns = 4 + cfg.MinConns = 0 + cfg.MaxConnLifetime = 30 * time.Minute + + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + return nil, fmt.Errorf("connect postgres: %w", err) + } + pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := pool.Ping(pingCtx); err != nil { + pool.Close() + return nil, fmt.Errorf("ping postgres: %w", err) + } + slog.Info("session version lookup enabled (postgres)") + return &Lookup{pool: pool}, nil +} + +func postgresConfigured() bool { + if os.Getenv("DATABASE_URL") != "" { + return true + } + return os.Getenv("PGHOST") != "" || os.Getenv("PGDATABASE") != "" +} + +// Close releases the pool. +func (l *Lookup) Close() { + if l != nil && l.pool != nil { + l.pool.Close() + } +} + +// LookupSessionVersion implements proxy.SessionVersionLookup. +func (l *Lookup) LookupSessionVersion(ctx context.Context, escrowID string) (string, bool, error) { + if l == nil || l.pool == nil { + return "", false, nil + } + if escrowID == "" { + return "", false, nil + } + qctx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + + var version *string + err := l.pool.QueryRow(qctx, ` + SELECT s.version + FROM devshard_session_index i + JOIN devshard_sessions s + ON s.epoch_id = i.epoch_id AND s.escrow_id = i.escrow_id + WHERE i.escrow_id = $1 + AND s.status = 'active' + LIMIT 1`, escrowID).Scan(&version) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return "", false, nil + } + return "", false, err + } + if version == nil || *version == "" { + return "", false, nil + } + return *version, true, nil +} From c1b91ea903cb231f6c3cfbefd517abc971a5cedf Mon Sep 17 00:00:00 2001 From: akup Date: Tue, 14 Jul 2026 22:03:03 +0300 Subject: [PATCH 2/5] fix(devshard): harden versionless obs routing and stats polling O(1) stats detail + negative cache, nginx obs rate limits, numeric primary version sort, single GetEscrow on owner bind, healthz not rewritten to versiond, and visible PG lookup fan-out errors. --- .../cmd/devshardd/session/bind_obs_test.go | 48 ++++++- devshard/cmd/devshardd/session/manager.go | 37 +++-- .../cmd/devshardd/session/manager_test.go | 4 +- devshard/cmd/devshardd/session/stats.go | 134 +++++++++--------- devshard/cmd/devshardd/session/stats_test.go | 80 +++++++++++ proxy/README.md | 14 +- proxy/entrypoint.sh | 84 +++++++++-- proxy/nginx.unified.conf.template | 1 + versioned/internal/proxy/lookup_telemetry.go | 44 ++++++ versioned/internal/proxy/proxy.go | 30 ++-- versioned/internal/proxy/proxy_test.go | 42 +++++- versioned/internal/proxy/version_sort.go | 125 ++++++++++++++++ versioned/internal/proxy/version_sort_test.go | 70 +++++++++ 13 files changed, 602 insertions(+), 111 deletions(-) create mode 100644 versioned/internal/proxy/lookup_telemetry.go create mode 100644 versioned/internal/proxy/version_sort.go create mode 100644 versioned/internal/proxy/version_sort_test.go diff --git a/devshard/cmd/devshardd/session/bind_obs_test.go b/devshard/cmd/devshardd/session/bind_obs_test.go index a1b4780759..891283a4f6 100644 --- a/devshard/cmd/devshardd/session/bind_obs_test.go +++ b/devshard/cmd/devshardd/session/bind_obs_test.go @@ -129,6 +129,52 @@ func TestOwnerChat_BindsSession(t *testing.T) { require.Equal(t, user.Address(), meta.CreatorAddr) } +type countingGetEscrowBridge struct { + bridge.MainnetBridge + calls int +} + +func (b *countingGetEscrowBridge) GetEscrow(escrowID string) (*bridge.EscrowInfo, error) { + b.calls++ + return b.MainnetBridge.GetEscrow(escrowID) +} + +func TestOwnerChat_FirstBindSingleGetEscrow(t *testing.T) { + const escrowID = "owner-bind-once" + store := newManagerTestStore(t) + hosts := make([]*signing.Secp256k1Signer, 3) + for i := range hosts { + hosts[i] = mustGenerateKey(t) + } + user := mustGenerateKey(t) + addresses := make([]string, len(hosts)) + for i, h := range hosts { + addresses[i] = h.Address() + } + inner := &mockBridge{ + escrow: &bridge.EscrowInfo{ + EscrowID: escrowID, + EpochID: 7, + Amount: 100000, + CreatorAddress: user.Address(), + Slots: addresses, + TokenPrice: 1, + }, + } + br := &countingGetEscrowBridge{MainnetBridge: inner} + mgr := NewHostManager(store, hosts[0], stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, br, nil, nil) + + e := echo.New() + mgr.Register(e.Group("")) + body := []byte(`{"model":"m","messages":[{"role":"user","content":"hi"}]}`) + _ = signedPOST(t, e, user, "/sessions/"+escrowID+"/chat/completions", escrowID, body) + + meta, err := store.GetSessionMeta(escrowID) + require.NoError(t, err) + require.Equal(t, testutil.RuntimeTestVersion, meta.Version) + require.Equal(t, 1, br.calls, "first-bind path must GetEscrow once (reuse for BuildGroup + CreateSession)") +} + func TestNonOwnerChat_DoesNotBindSession(t *testing.T) { const escrowID = "non-owner-bind" mgr, store, _, hostSigner := setupBindTestManager(t, escrowID) @@ -178,7 +224,7 @@ func TestGetOrCreate_RecoversBeforeCreate(t *testing.T) { } mgr := NewHostManager(store, hostSigner, stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, br, nil, nil) - srv, err := mgr.getOrCreate("1") + srv, err := mgr.getOrCreate("1", nil) require.NoError(t, err) require.Equal(t, uint64(2), srv.Host().SnapshotState().LatestNonce) } diff --git a/devshard/cmd/devshardd/session/manager.go b/devshard/cmd/devshardd/session/manager.go index dcb96aa0a0..22c16ca457 100644 --- a/devshard/cmd/devshardd/session/manager.go +++ b/devshard/cmd/devshardd/session/manager.go @@ -56,10 +56,11 @@ type HostManager struct { availability devshardpkg.AvailabilityProvider maxNonce devshardpkg.MaxNonceProvider - statsMu sync.Mutex - statsShardsCache *statsShardsResponse - statsShardsCached time.Time - statsDetailsCache map[string]statsShardDetailCache + statsMu sync.Mutex + statsShardsCache *statsShardsResponse + statsShardsCached time.Time + statsDetailsCache map[string]statsShardDetailCache + statsNegativeCache map[string]statsNegativeCacheEntry } const ( @@ -98,6 +99,7 @@ func NewHostManager( payloadStore: ps, recorder: recorder, statsDetailsCache: make(map[string]statsShardDetailCache), + statsNegativeCache: make(map[string]statsNegativeCacheEntry), } } @@ -130,7 +132,7 @@ func (m *HostManager) Close() error { // SessionServer resolves or creates the per-escrow transport server. // Prefer BindOwnerChat for HTTP chat; this remains for tests and explicit bind. func (m *HostManager) SessionServer(escrowID string) (*transport.Server, error) { - return m.getOrCreate(escrowID) + return m.getOrCreate(escrowID, nil) } // SessionServerExisting returns a live or recovered session server. @@ -177,7 +179,8 @@ func (m *HostManager) BindOwnerChat(c echo.Context) (*transport.Server, error) { return nil, echo.NewHTTPError(http.StatusForbidden, "restricted to escrow owner") } - srv, err = m.getOrCreate(escrowID) + // Pass the already-fetched escrow so create() does not GetEscrow again. + srv, err = m.getOrCreate(escrowID, escrow) if err != nil { return nil, err } @@ -209,7 +212,10 @@ func (m *HostManager) HandleSettlementFinalized(escrowID string) error { return nil } -func (m *HostManager) getOrCreate(escrowID string) (*transport.Server, error) { +// getOrCreate returns a live session, recovering from store or creating. +// When escrow is non-nil (BindOwnerChat first-bind path), create reuses it and +// skips a second bridge.GetEscrow. +func (m *HostManager) getOrCreate(escrowID string, escrow *bridge.EscrowInfo) (*transport.Server, error) { if srv, ok := m.existingServer(escrowID); ok { return srv, nil } @@ -234,7 +240,7 @@ func (m *HostManager) getOrCreate(escrowID string) (*transport.Server, error) { return nil, err } - srv, err = m.create(escrowID) + srv, err = m.create(escrowID, escrow) if err != nil { return nil, err } @@ -330,15 +336,18 @@ func (m *HostManager) EvictBefore(cutoffEpoch uint64) int { return len(evicted) } -func (m *HostManager) create(escrowID string) (*transport.Server, error) { - group, err := bridge.BuildGroup(escrowID, m.bridge) - if err != nil { - return nil, fmt.Errorf("build group: %w", err) +func (m *HostManager) create(escrowID string, escrow *bridge.EscrowInfo) (*transport.Server, error) { + if escrow == nil { + var err error + escrow, err = m.bridge.GetEscrow(escrowID) + if err != nil { + return nil, fmt.Errorf("get escrow: %w", err) + } } - escrow, err := m.bridge.GetEscrow(escrowID) + group, err := bridge.BuildGroupFromEscrow(escrow) if err != nil { - return nil, fmt.Errorf("get escrow: %w", err) + return nil, fmt.Errorf("build group: %w", err) } creatorAddr := escrow.CreatorAddress diff --git a/devshard/cmd/devshardd/session/manager_test.go b/devshard/cmd/devshardd/session/manager_test.go index d86734862b..d6f33e31bd 100644 --- a/devshard/cmd/devshardd/session/manager_test.go +++ b/devshard/cmd/devshardd/session/manager_test.go @@ -219,7 +219,7 @@ func TestRecoverSessions_Nonce0(t *testing.T) { require.NotNil(t, srv) require.NotNil(t, srv.Host()) - srv2, err := mgr.getOrCreate("1") + srv2, err := mgr.getOrCreate("1", nil) require.NoError(t, err) require.Equal(t, srv, srv2) } @@ -248,7 +248,7 @@ func TestCreateSession_BindsConfiguredVersion(t *testing.T) { const standaloneVersion = "v0.2.11" mgr := NewHostManager(store, hosts[0], stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, standaloneVersion, br, nil, nil) - _, err := mgr.getOrCreate("1") + _, err := mgr.getOrCreate("1", nil) require.NoError(t, err) meta, err := store.GetSessionMeta("1") diff --git a/devshard/cmd/devshardd/session/stats.go b/devshard/cmd/devshardd/session/stats.go index ec00a406a2..cd84713339 100644 --- a/devshard/cmd/devshardd/session/stats.go +++ b/devshard/cmd/devshardd/session/stats.go @@ -20,13 +20,22 @@ import ( "devshard/types" ) -const statsCacheTTL = 60 * time.Second +const ( + statsCacheTTL = 60 * time.Second + statsNegativeCacheTTL = 10 * time.Second + statsNegativeCacheMax = 4096 +) type statsShardDetailCache struct { response *statsShardDetailResponse cached time.Time } +type statsNegativeCacheEntry struct { + reason string + cached time.Time +} + type statsShardsResponse struct { CurrentEpochID uint64 `json:"current_epoch_id"` CachedAt int64 `json:"cached_at"` @@ -110,6 +119,9 @@ func recordStatsSessionResolutionFailure(c echo.Context, escrowID string, err er } func statsSessionResolutionStatus(err error) (observability.MetricStatus, observability.Reason) { + if errors.Is(err, storage.ErrSessionNotFound) { + return observability.MetricStatusError, observability.ReasonSessionResolveErr + } if errors.Is(err, storage.ErrSessionVersionConflict) { return observability.MetricStatusError, observability.ReasonVersionConflict } @@ -172,14 +184,26 @@ func (m *HostManager) statsShardDetail(escrowID string, now time.Time) (*statsSh m.statsMu.Unlock() return resp, nil } + if neg, ok := m.statsNegativeCache[escrowID]; ok && now.Sub(neg.cached) < statsNegativeCacheTTL { + m.statsMu.Unlock() + return nil, storage.ErrSessionNotFound + } m.statsMu.Unlock() - sess, err := m.boundVersionActiveSession(escrowID) + sess, reason, metaVersion, err := m.lookupBoundVersionActiveSession(escrowID) if err != nil { + if errors.Is(err, storage.ErrSessionNotFound) { + m.rememberStatsMiss(escrowID, reason, metaVersion, now) + } return nil, err } + m.clearStatsMiss(escrowID) + srv, err := m.SessionServerExisting(escrowID) if err != nil { + if errors.Is(err, storage.ErrSessionNotFound) { + m.rememberStatsMiss(escrowID, "absent_from_memory_store", metaVersion, now) + } return nil, err } @@ -200,93 +224,75 @@ func (m *HostManager) statsShardDetail(escrowID string, now time.Time) (*statsSh m.statsMu.Lock() m.statsDetailsCache[escrowID] = statsShardDetailCache{response: resp, cached: now} + delete(m.statsNegativeCache, escrowID) m.statsMu.Unlock() return resp, nil } -func (m *HostManager) boundVersionActiveSession(escrowID string) (storage.ActiveSession, error) { - currentEpochID, active, err := m.boundVersionActiveSessions() +// lookupBoundVersionActiveSession is O(1): one GetSessionMeta, no full-store scan. +// reason/metaVersion are set on not-found for negative-cache logging. +func (m *HostManager) lookupBoundVersionActiveSession(escrowID string) (storage.ActiveSession, string, string, error) { + meta, err := m.store.GetSessionMeta(escrowID) if err != nil { - return storage.ActiveSession{}, err - } - for _, sess := range active { - if sess.EscrowID == escrowID { - return sess, nil + if errors.Is(err, storage.ErrSessionNotFound) { + return storage.ActiveSession{}, "absent_from_active_store", "", storage.ErrSessionNotFound } + return storage.ActiveSession{}, "meta_unreadable", "", err } - m.logStatsSessionNotFound(escrowID, currentEpochID, active) - return storage.ActiveSession{}, storage.ErrSessionNotFound -} - -// logStatsSessionNotFound explains why detail stats 404'd: missing from storage -// (pruned / never created) or filtered by bound version. Rate-limited only by -// the caller's polling; keep fields compact for grep in CI dumps. -func (m *HostManager) logStatsSessionNotFound(escrowID string, currentEpochID uint64, filtered []storage.ActiveSession) { - all, listErr := m.store.ListActiveSessions() - filteredIDs := make([]string, 0, len(filtered)) - for _, sess := range filtered { - filteredIDs = append(filteredIDs, sess.EscrowID) - } - - var ( - foundInStore bool - sessionEpoch uint64 - metaVersion string - versionMatch bool - metaErr error - ) - if listErr == nil { - for _, sess := range all { - if sess.EscrowID != escrowID { - continue - } - foundInStore = true - sessionEpoch = sess.EpochID - metaVersion, versionMatch, metaErr = m.sessionMatchesBoundVersion(escrowID) - break - } + if meta.Version != "" && meta.Version != m.boundVersion { + return storage.ActiveSession{}, "version_mismatch", meta.Version, storage.ErrSessionNotFound } + return storage.ActiveSession{EscrowID: escrowID, EpochID: meta.EpochID}, "", meta.Version, nil +} - reason := "absent_from_active_store" - switch { - case listErr != nil: - reason = "list_active_failed" - case !foundInStore: - reason = "absent_from_active_store" - case metaErr != nil: - reason = "meta_unreadable" - case !versionMatch: - reason = "version_mismatch" - default: - reason = "filtered_unknown" +func (m *HostManager) rememberStatsMiss(escrowID, reason, metaVersion string, now time.Time) { + m.statsMu.Lock() + if neg, ok := m.statsNegativeCache[escrowID]; ok && now.Sub(neg.cached) < statsNegativeCacheTTL { + m.statsMu.Unlock() + return } - - storeSummary := make([]string, 0, len(all)) - for _, sess := range all { - storeSummary = append(storeSummary, fmt.Sprintf("%s@%d", sess.EscrowID, sess.EpochID)) + if len(m.statsNegativeCache) >= statsNegativeCacheMax { + m.pruneStatsNegativeCacheLocked(now) } + m.statsNegativeCache[escrowID] = statsNegativeCacheEntry{reason: reason, cached: now} + m.statsMu.Unlock() + // Log once per escrow per negative-cache TTL (edge-triggered on insert). logging.Warn("devshard stats session not in bound-version active set", inferenceTypes.System, "escrow_id", escrowID, "filter_reason", reason, - "current_epoch_id", currentEpochID, - "session_epoch_id", sessionEpoch, - "found_in_store", foundInStore, "bound_version", m.boundVersion, "meta_version", metaVersion, - "version_match", versionMatch, - "meta_error", metaErr, - "list_error", listErr, - "filtered_escrows", strings.Join(filteredIDs, ","), - "store_escrows", strings.Join(storeSummary, ","), ) } +func (m *HostManager) clearStatsMiss(escrowID string) { + m.statsMu.Lock() + delete(m.statsNegativeCache, escrowID) + m.statsMu.Unlock() +} + +func (m *HostManager) pruneStatsNegativeCacheLocked(now time.Time) { + for id, neg := range m.statsNegativeCache { + if now.Sub(neg.cached) >= statsNegativeCacheTTL { + delete(m.statsNegativeCache, id) + } + } + // Still over cap: drop arbitrary entries until under limit. + for id := range m.statsNegativeCache { + if len(m.statsNegativeCache) < statsNegativeCacheMax { + break + } + delete(m.statsNegativeCache, id) + } +} + // boundVersionActiveSessions returns non-pruned active sessions for this // versiond bind. Epoch is recorded on each shard but not used as a filter: // unsettled sessions remain queryable across epoch boundaries until prune. // currentEpochID is still returned for list metadata / operators. +// Used by the list endpoint only; detail uses lookupBoundVersionActiveSession. func (m *HostManager) boundVersionActiveSessions() (uint64, []storage.ActiveSession, error) { active, err := m.store.ListActiveSessions() if err != nil { diff --git a/devshard/cmd/devshardd/session/stats_test.go b/devshard/cmd/devshardd/session/stats_test.go index 99f3c93ead..b305779b84 100644 --- a/devshard/cmd/devshardd/session/stats_test.go +++ b/devshard/cmd/devshardd/session/stats_test.go @@ -292,3 +292,83 @@ func TestStatsShardDetailSkipsForeignVersionSession(t *testing.T) { rec := requestStats(t, mgr, "/v1/devshard", "/stats/shards/escrow-v2") require.Equal(t, http.StatusNotFound, rec.Code, "body: %s", rec.Body.String()) } + +type countingMetaStore struct { + storage.Storage + metaCalls int + listCalls int +} + +func (s *countingMetaStore) GetSessionMeta(escrowID string) (*storage.SessionMeta, error) { + s.metaCalls++ + return s.Storage.GetSessionMeta(escrowID) +} + +func (s *countingMetaStore) ListActiveSessions() ([]storage.ActiveSession, error) { + s.listCalls++ + return s.Storage.ListActiveSessions() +} + +func TestStatsShardDetailResolvesO1WithoutListScan(t *testing.T) { + base := newManagerTestStore(t) + group, user, hostSigner := createStoredSession(t, base, "escrow-o1", 7, 1) + createStoredSession(t, base, "escrow-noise-a", 7, 0) + createStoredSession(t, base, "escrow-noise-b", 7, 0) + + counting := &countingMetaStore{Storage: currentEpochStore{Storage: base, epoch: 7}} + addresses := make([]string, len(group)) + for i, s := range group { + addresses[i] = s.ValidatorAddress + } + br := &mockBridge{ + escrow: &bridge.EscrowInfo{ + EscrowID: "escrow-o1", + EpochID: 7, + Amount: 100000000, + CreatorAddress: user.Address(), + Slots: addresses, + }, + } + mgr := NewHostManager(counting, hostSigner, stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, br, nil, nil) + require.NoError(t, mgr.RecoverSessions()) + counting.metaCalls = 0 + counting.listCalls = 0 + + rec := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/escrow-o1") + require.Equal(t, http.StatusOK, rec.Code, "body: %s", rec.Body.String()) + require.Equal(t, 0, counting.listCalls, "detail must not ListActiveSessions") + require.Equal(t, 1, counting.metaCalls, "detail should GetSessionMeta once") +} + +func TestStatsShardDetailNegativeCacheSkipsStoreOnRepeatMiss(t *testing.T) { + base := newManagerTestStore(t) + _, _, hostSigner := createStoredSession(t, base, "escrow-other", 7, 0) + counting := &countingMetaStore{Storage: currentEpochStore{Storage: base, epoch: 7}} + mgr := NewHostManager(counting, hostSigner, stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, &mockBridge{}, nil, nil) + + rec1 := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/missing-escrow") + require.Equal(t, http.StatusNotFound, rec1.Code, "body: %s", rec1.Body.String()) + require.Equal(t, 0, counting.listCalls) + require.Equal(t, 1, counting.metaCalls) + + rec2 := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/missing-escrow") + require.Equal(t, http.StatusNotFound, rec2.Code, "body: %s", rec2.Body.String()) + require.Equal(t, 0, counting.listCalls) + require.Equal(t, 1, counting.metaCalls, "negative cache must skip GetSessionMeta on repeat miss") +} + +func TestStatsShardDetailNegativeCacheForVersionMismatch(t *testing.T) { + base := newManagerTestStore(t) + _, _, hostSigner := createStoredSessionWithVersion(t, base, "escrow-foreign", 7, "foreign", 0) + counting := &countingMetaStore{Storage: currentEpochStore{Storage: base, epoch: 7}} + mgr := NewHostManager(counting, hostSigner, stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, &mockBridge{}, nil, nil) + + rec1 := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/escrow-foreign") + require.Equal(t, http.StatusNotFound, rec1.Code) + require.Equal(t, 1, counting.metaCalls) + require.Equal(t, 0, counting.listCalls) + + rec2 := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/escrow-foreign") + require.Equal(t, http.StatusNotFound, rec2.Code) + require.Equal(t, 1, counting.metaCalls, "version mismatch should be negatively cached") +} diff --git a/proxy/README.md b/proxy/README.md index 728a82a091..e874a54ba1 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -114,6 +114,9 @@ Key runtime environment variables: | `VERSIOND_SERVICE_NAME` | versiond | Upstream for `/devshard/` (and legacy `/v1/devshard/` after rewrite). Set to `versiond-router` for sticky multi-versiond overlay. | | `VERSIOND_PORT` | 8080 | Port on the versiond (or versiond-router) upstream. | | `DISABLE_DEVSHARD_PROXY` | false | Set to `true` to disable `/devshard/` and `/v1/devshard/` routing to versiond. | +| `DEVSHARD_OBS_RATE_LIMIT_RPS` | 10 | Per-IP rate limit for public observability GETs (`/devshard/sessions|stats|metrics|healthz` and rewritten legacy obs URLs). Protocol chat/gossip/payloads stay on the exempt zone. | +| `DEVSHARD_OBS_RATE_UNIT` | s | Unit for obs rate (`s` or `m`). | +| `DEVSHARD_OBS_BURST` | 20 | Burst for obs rate limit. | Versiond-side (on the versiond container, not the join proxy): @@ -138,6 +141,8 @@ GET /devshard/sessions/{escrow_id}/signatures GET /devshard/stats/shards GET /devshard/stats/shards/{escrow_id} GET /devshard/metrics +GET /devshard/healthz # versiond supervisor (not a child) +GET /devshard/{version}/healthz # that child's healthz ``` | Client URL (legacy, still works) | Internal route to versiond | @@ -147,14 +152,17 @@ GET /devshard/metrics | `GET /devshard/{version}/sessions/{id}/signatures` | `/devshard/sessions/{id}/signatures` | | `GET /devshard/{version}/stats/shards…` | `/devshard/stats/shards…` | | `GET /devshard/{version}/metrics` | `/devshard/metrics` | -| `GET /devshard/{version}/healthz` | `/devshard/healthz` | +| `GET /devshard/{version}/healthz` | **not rewritten** — proxied as `/{version}/healthz` to that child | Protocol traffic stays versioned: `POST …/chat/completions`, gossip, challenge-receipt, and `GET …/payloads` are **not** rewritten. +Public obs paths (versionless and rewritten legacy) use a dedicated nginx zone (`devshard_obs`, default `10r/s` burst `20`) so scrapers cannot amplify polling under the exempt chat limits. Chat / gossip / payloads remain on the exempt catch-all. + versiond serves the versionless obs paths: -- **Session-scoped** (`/sessions/{id}/diffs|mempool|signatures`, `/stats/shards/{id}`): when Postgres is configured (`PGHOST` / `DATABASE_URL`), route by `sessions.version`; unbound → 404. If lookup is disabled or PG errors, fan-out across children. -- **Process-level** (`/metrics`, `/healthz`, `/stats/shards` list): pin to lexicographic-max running version (list is not merged across versions). +- **Session-scoped** (`/sessions/{id}/diffs|mempool|signatures`, `/stats/shards/{id}`): when Postgres is configured (`PGHOST` / `DATABASE_URL`), route by `sessions.version`; unbound → 404. If lookup is disabled or PG errors, fan-out across children. Lookup errors emit a rate-limited warn (`session version lookup failed; falling back to fan-out`) and increment an in-process counter (`proxy.LookupFanoutErrors`) — they are not silent. +- **Process-level** (`/metrics`, `/stats/shards` list): pin to newest running version by numeric/dotted comparison (`v10` > `v2`, `v0.2.11` > `v0.2.9`), not lexicographic order. +- **Health:** `GET /healthz` on versiond is **supervisor** status (mux, ahead of the proxy). Join proxy `GET /devshard/healthz` hits that. Per-child health is `GET /devshard/{version}/healthz` (not rewritten). Do not use versionless `/healthz` to probe a specific child. Disable lookup: `VERSIOND_DISABLE_SESSION_LOOKUP=true`. diff --git a/proxy/entrypoint.sh b/proxy/entrypoint.sh index 6f57e31bb3..a28fee9088 100755 --- a/proxy/entrypoint.sh +++ b/proxy/entrypoint.sh @@ -510,6 +510,13 @@ CHAIN_API_EXEMPT_ROUTES=${CHAIN_API_EXEMPT_ROUTES:-""} CHAIN_RPC_EXEMPT_ROUTES=${CHAIN_RPC_EXEMPT_ROUTES:-""} CHAIN_GRPC_EXEMPT_ROUTES=${CHAIN_GRPC_EXEMPT_ROUTES:-""} +# Public DevShard observability (unauthenticated GETs). Tighter than exempt so +# scrapers cannot amplify stats/diffs polling. Protocol (chat/gossip/payloads) +# stays on the exempt catch-all under /devshard/. +DEVSHARD_OBS_RATE_LIMIT_VAL=${DEVSHARD_OBS_RATE_LIMIT_RPS:-10} +DEVSHARD_OBS_RATE_UNIT=${DEVSHARD_OBS_RATE_UNIT:-s} +DEVSHARD_OBS_BURST=${DEVSHARD_OBS_BURST:-20} + # Chain API CHAIN_API_RATE_LIMIT_VAL=${CHAIN_API_RATE_LIMIT_RPS:-20} CHAIN_API_RATE_UNIT=${CHAIN_API_RATE_UNIT:-m} @@ -542,6 +549,7 @@ echo "Rate Limits:" echo " Global: ${GLOBAL_RATE_LIMIT_VAL}r/${GLOBAL_RATE_UNIT} (burst=${GLOBAL_BURST})" echo " App API (Standard): ${GONKA_API_RATE_LIMIT_VAL}r/${GONKA_API_RATE_UNIT} (burst=${GONKA_API_BURST})" echo " App API (Exempt): ${EXEMPT_RATE_LIMIT_VAL}r/${EXEMPT_RATE_UNIT} (burst=${EXEMPT_BURST}) -> [${GONKA_API_EXEMPT_ROUTES}]" +echo " DevShard obs: ${DEVSHARD_OBS_RATE_LIMIT_VAL}r/${DEVSHARD_OBS_RATE_UNIT} (burst=${DEVSHARD_OBS_BURST})" echo " Chain API: ${CHAIN_API_RATE_LIMIT_VAL}r/${CHAIN_API_RATE_UNIT} (burst=${CHAIN_API_BURST})" echo " Chain RPC: ${CHAIN_RPC_RATE_LIMIT_VAL}r/${CHAIN_RPC_RATE_UNIT} (burst=${CHAIN_RPC_BURST})" echo " Chain gRPC: ${CHAIN_GRPC_RATE_LIMIT_VAL}r/${CHAIN_GRPC_RATE_UNIT} (burst=${CHAIN_GRPC_BURST})" @@ -556,6 +564,7 @@ echo " Chain gRPC: [${CHAIN_GRPC_BLOCKED_ROUTES}]" export LIMIT_REQ_ZONE_GLOBAL="limit_req_zone \$\$whitelist_limit_key zone=global_zone:10m rate=${GLOBAL_RATE_LIMIT_VAL}r/${GLOBAL_RATE_UNIT};" export LIMIT_REQ_ZONE_GONKA_API="limit_req_zone \$\$whitelist_limit_key zone=api_zone:10m rate=${GONKA_API_RATE_LIMIT_VAL}r/${GONKA_API_RATE_UNIT};" export LIMIT_REQ_ZONE_EXEMPT="limit_req_zone \$\$whitelist_limit_key zone=exempt_zone:10m rate=${EXEMPT_RATE_LIMIT_VAL}r/${EXEMPT_RATE_UNIT};" +export LIMIT_REQ_ZONE_DEVSHARD_OBS="limit_req_zone \$\$whitelist_limit_key zone=devshard_obs:10m rate=${DEVSHARD_OBS_RATE_LIMIT_VAL}r/${DEVSHARD_OBS_RATE_UNIT};" export LIMIT_REQ_ZONE_CHAIN_API="limit_req_zone \$\$whitelist_limit_key zone=chain_api_zone:10m rate=${CHAIN_API_RATE_LIMIT_VAL}r/${CHAIN_API_RATE_UNIT};" export LIMIT_REQ_ZONE_CHAIN_RPC="limit_req_zone \$\$whitelist_limit_key zone=rpc_zone:10m rate=${CHAIN_RPC_RATE_LIMIT_VAL}r/${CHAIN_RPC_RATE_UNIT};" export LIMIT_REQ_ZONE_CHAIN_GRPC="limit_req_zone \$\$whitelist_limit_key zone=grpc_zone:10m rate=${CHAIN_GRPC_RATE_LIMIT_VAL}r/${CHAIN_GRPC_RATE_UNIT};" @@ -600,14 +609,16 @@ else fi # /devshard/ location -- forwards to versiond which dispatches to the matching -# child binary. Treated as exempt (inference forwarding): streaming, long -# timeouts, exempt rate/conn limits, CORS. +# child binary. # # Phase 1: versioned observability paths are rewritten internally to versionless # canonical URLs (no client-visible redirect). Dashboards that hardcode # /devshard/{version}/sessions/.../diffs keep working; the version segment is -# dropped before versiond so it cannot participate in protocol bind. Protocol -# POSTs and payloads stay on /devshard/{version}/... via the catch-all below. +# dropped before versiond so it cannot participate in protocol bind. +# +# Public obs (versionless + rewritten legacy) uses a tighter rate-limit zone. +# Protocol POSTs and payloads stay on /devshard/{version}/... via the exempt +# catch-all below (chat/SSE need high limits). if [ "${DISABLE_DEVSHARD_PROXY}" != "true" ]; then export DEVSHARD_VERSIOND_LOCATION="# Versioned obs → versionless (internal rewrite); protocol stays versioned location ~ ^/devshard/[^/]+/sessions/([^/]+)/(diffs|mempool|signatures)\$ { @@ -616,8 +627,62 @@ if [ "${DISABLE_DEVSHARD_PROXY}" != "true" ]; then location ~ ^/devshard/[^/]+/stats/shards(/.*)?\$ { rewrite ^ /devshard/stats/shards\$\$1 last; } - location ~ ^/devshard/[^/]+/(metrics|healthz)\$ { - rewrite ^ /devshard/\$\$1 last; + location ~ ^/devshard/[^/]+/metrics\$ { + rewrite ^ /devshard/metrics last; + } + # /devshard/{version}/healthz is NOT rewritten — it must reach that child. + # Versionless /devshard/healthz is versiond's own supervisor health (mux). + # Versionless public observability — tighter than exempt protocol limits + location ~ ^/devshard/sessions/[^/]+/(diffs|mempool|signatures)\$ { + set \$limit_zone_name \"DEVSHARD_OBS\"; + limit_req zone=devshard_obs burst=${DEVSHARD_OBS_BURST} nodelay; + ${LIMIT_CONN_RULE_EXEMPT} + rewrite ^/devshard/(.*)\$ /\$\$1 break; + proxy_pass http://versiond_backend; + proxy_set_header Host \$\$host; + proxy_set_header X-Real-IP \$\$remote_addr; + proxy_set_header X-Forwarded-For \$\$proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto \$\$scheme; + proxy_set_header Authorization \$\$http_authorization; + ${CORS_CONFIG} + ${STREAMING_CONFIG} + proxy_connect_timeout ${GONKA_API_CONNECT_TIMEOUT}s; + proxy_send_timeout ${GONKA_API_TRANSFER_TIMEOUT}s; + proxy_read_timeout ${GONKA_API_TRANSFER_TIMEOUT}s; + } + location ~ ^/devshard/stats/ { + set \$limit_zone_name \"DEVSHARD_OBS\"; + limit_req zone=devshard_obs burst=${DEVSHARD_OBS_BURST} nodelay; + ${LIMIT_CONN_RULE_EXEMPT} + rewrite ^/devshard/(.*)\$ /\$\$1 break; + proxy_pass http://versiond_backend; + proxy_set_header Host \$\$host; + proxy_set_header X-Real-IP \$\$remote_addr; + proxy_set_header X-Forwarded-For \$\$proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto \$\$scheme; + proxy_set_header Authorization \$\$http_authorization; + ${CORS_CONFIG} + ${STREAMING_CONFIG} + proxy_connect_timeout ${GONKA_API_CONNECT_TIMEOUT}s; + proxy_send_timeout ${GONKA_API_TRANSFER_TIMEOUT}s; + proxy_read_timeout ${GONKA_API_TRANSFER_TIMEOUT}s; + } + location ~ ^/devshard/(metrics|healthz)\$ { + set \$limit_zone_name \"DEVSHARD_OBS\"; + limit_req zone=devshard_obs burst=${DEVSHARD_OBS_BURST} nodelay; + ${LIMIT_CONN_RULE_EXEMPT} + rewrite ^/devshard/(.*)\$ /\$\$1 break; + proxy_pass http://versiond_backend; + proxy_set_header Host \$\$host; + proxy_set_header X-Real-IP \$\$remote_addr; + proxy_set_header X-Forwarded-For \$\$proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto \$\$scheme; + proxy_set_header Authorization \$\$http_authorization; + ${CORS_CONFIG} + ${STREAMING_CONFIG} + proxy_connect_timeout ${GONKA_API_CONNECT_TIMEOUT}s; + proxy_send_timeout ${GONKA_API_TRANSFER_TIMEOUT}s; + proxy_read_timeout ${GONKA_API_TRANSFER_TIMEOUT}s; } location /devshard/ { set \$limit_zone_name \"EXEMPT\"; @@ -640,6 +705,7 @@ if [ "${DISABLE_DEVSHARD_PROXY}" != "true" ]; then }" else export DEVSHARD_VERSIOND_LOCATION="# devshard proxy disabled" + export LIMIT_REQ_ZONE_DEVSHARD_OBS="" fi # -------------------------------------------------------------------------------- @@ -1144,7 +1210,7 @@ ENVSUBST_VARS="${ENVSUBST_VARS},\$JAEGER_PORT,\$JAEGER_BASE_PATH,\$JAEGER_UPSTRE ENVSUBST_VARS="${ENVSUBST_VARS},\$GRAFANA_PORT,\$GRAFANA_BASE_PATH,\$GRAFANA_UPSTREAM,\$GRAFANA_LOCATION" # Group 5: Rate Limiting Zones -ENVSUBST_VARS="${ENVSUBST_VARS},\$LIMIT_REQ_ZONE_GLOBAL,\$LIMIT_REQ_ZONE_GONKA_API,\$LIMIT_REQ_ZONE_EXEMPT" +ENVSUBST_VARS="${ENVSUBST_VARS},\$LIMIT_REQ_ZONE_GLOBAL,\$LIMIT_REQ_ZONE_GONKA_API,\$LIMIT_REQ_ZONE_EXEMPT,\$LIMIT_REQ_ZONE_DEVSHARD_OBS" ENVSUBST_VARS="${ENVSUBST_VARS},\$LIMIT_REQ_ZONE_CHAIN_RPC,\$LIMIT_REQ_ZONE_CHAIN_API,\$LIMIT_REQ_ZONE_CHAIN_GRPC" # Group 5b: Concurrency Zones and Rules @@ -1218,7 +1284,9 @@ fi if [ "${DISABLE_DEVSHARD_PROXY}" != "true" ]; then echo " /devshard/* -> Versiond (devshard binaries)" echo " /devshard/{v}/sessions/*/diffs|mempool|signatures -> rewrite /devshard/sessions/..." - echo " /devshard/{v}/stats/* /metrics /healthz -> rewrite versionless (internal)" + echo " /devshard/{v}/stats/* /metrics -> rewrite versionless (internal)" + echo " /devshard/{v}/healthz -> child (not rewritten); /devshard/healthz -> versiond" + echo " /devshard/sessions|stats|metrics|healthz -> obs rate limit ${DEVSHARD_OBS_RATE_LIMIT_VAL}r/${DEVSHARD_OBS_RATE_UNIT}" echo " /v1/devshard/* -> /devshard/v1/* (legacy rewrite)" fi echo " /health -> Health check" diff --git a/proxy/nginx.unified.conf.template b/proxy/nginx.unified.conf.template index 03e6a19df4..11dad3363c 100644 --- a/proxy/nginx.unified.conf.template +++ b/proxy/nginx.unified.conf.template @@ -6,6 +6,7 @@ http { ${LIMIT_REQ_ZONE_GLOBAL} ${LIMIT_REQ_ZONE_EXEMPT} ${LIMIT_REQ_ZONE_GONKA_API} + ${LIMIT_REQ_ZONE_DEVSHARD_OBS} ${LIMIT_REQ_ZONE_CHAIN_RPC} ${LIMIT_REQ_ZONE_CHAIN_API} ${LIMIT_REQ_ZONE_CHAIN_GRPC} diff --git a/versioned/internal/proxy/lookup_telemetry.go b/versioned/internal/proxy/lookup_telemetry.go new file mode 100644 index 0000000000..0e8d618a3c --- /dev/null +++ b/versioned/internal/proxy/lookup_telemetry.go @@ -0,0 +1,44 @@ +package proxy + +import ( + "log/slog" + "sync/atomic" + "time" +) + +// lookupFanoutErrors counts session-version lookup failures that fell back to +// fan-out. Exposed for tests and ops (also included in rate-limited warn logs). +var lookupFanoutErrors atomic.Uint64 + +// lookupWarnLastUnixNano gates warn spam when PG is persistently broken. +var lookupWarnLastUnixNano atomic.Int64 + +const lookupWarnInterval = 30 * time.Second + +// LookupFanoutErrors returns how many times versionless obs degraded from PG +// lookup failure to fan-out since process start. +func LookupFanoutErrors() uint64 { + return lookupFanoutErrors.Load() +} + +func noteLookupFanout(escrowID string, err error) { + n := lookupFanoutErrors.Add(1) + now := time.Now().UnixNano() + last := lookupWarnLastUnixNano.Load() + if now-last < lookupWarnInterval.Nanoseconds() { + return + } + if !lookupWarnLastUnixNano.CompareAndSwap(last, now) { + return + } + slog.Warn("session version lookup failed; falling back to fan-out", + "escrow_id", escrowID, + "error", err, + "fanout_errors_total", n, + ) +} + +func resetLookupFanoutTelemetryForTest() { + lookupFanoutErrors.Store(0) + lookupWarnLastUnixNano.Store(0) +} diff --git a/versioned/internal/proxy/proxy.go b/versioned/internal/proxy/proxy.go index 4de1eea125..cd8d4adddb 100644 --- a/versioned/internal/proxy/proxy.go +++ b/versioned/internal/proxy/proxy.go @@ -7,7 +7,6 @@ import ( "net/http/httptest" "net/http/httputil" "net/url" - "sort" "strings" "sync/atomic" ) @@ -38,8 +37,10 @@ func WithSessionVersionLookup(lookup SessionVersionLookup) HandlerOption { // Example: /v0.2.11/chat/completions -> localhost:9001/chat/completions // // Versionless observability paths (sessions/.../diffs|mempool|signatures, -// stats/..., metrics, healthz) are forwarded without a version prefix so join +// stats/..., metrics) are forwarded without a version prefix so join // proxy can rewrite legacy /{version}/obs URLs onto canonical versionless URLs. +// /healthz is intentionally NOT versionless here: versiond registers its own +// /healthz on the mux (supervisor status). Child health is /{version}/healthz. func Handler(routes *atomic.Value, opts ...HandlerOption) http.Handler { var cfg handlerConfig for _, opt := range opts { @@ -80,7 +81,7 @@ func Handler(routes *atomic.Value, opts ...HandlerOption) http.Handler { // a protocol version segment. payloads stays versioned (validator protocol). func isVersionlessObsPath(path string) bool { switch path { - case "metrics", "healthz": + case "metrics": return true } if path == "stats" || strings.HasPrefix(path, "stats/") { @@ -103,19 +104,21 @@ func serveVersionlessObs(w http.ResponseWriter, r *http.Request, routeMap map[st return } - // Process-level obs (/metrics, /healthz, /stats/shards list): pin to primary. + // Process-level obs (/metrics, /stats/shards list): pin to primary. // Multi-version aggregation of /stats/shards is deferred; primary is the - // lexicographic-max approved version (documented limitation for non-HA / multi-version lists). + // newest approved version by numeric/dotted comparison (not lexicographic). + // /healthz is owned by versiond's mux (not this handler). escrowID, scoped := escrowIDFromObsPath(rest) if !scoped { - reverseProxy(routeMap[versions[len(versions)-1]], rest).ServeHTTP(w, r) + reverseProxy(routeMap[primaryVersion(versions)], rest).ServeHTTP(w, r) return } if lookup != nil { ver, ok, err := lookup.LookupSessionVersion(r.Context(), escrowID) if err != nil { - // Shared index unavailable — degrade to fan-out. + // Shared index unavailable — degrade to fan-out (visible via warn + counter). + noteLookupFanout(escrowID, err) serveSessionObsFanout(w, r, routeMap, versions, rest) return } @@ -137,7 +140,9 @@ func serveVersionlessObs(w http.ResponseWriter, r *http.Request, routeMap map[st func serveSessionObsFanout(w http.ResponseWriter, r *http.Request, routeMap map[string]string, versions []string, rest string) { var fallback409 *httptest.ResponseRecorder - for _, ver := range versions { + // Try newest first — more likely to own recently bound escrows. + for i := len(versions) - 1; i >= 0; i-- { + ver := versions[i] rec := httptest.NewRecorder() reverseProxy(routeMap[ver], rest).ServeHTTP(rec, r.Clone(r.Context())) switch { @@ -178,15 +183,6 @@ func escrowIDFromObsPath(rest string) (string, bool) { return "", false } -func sortedVersions(routeMap map[string]string) []string { - versions := make([]string, 0, len(routeMap)) - for v := range routeMap { - versions = append(versions, v) - } - sort.Strings(versions) - return versions -} - func reverseProxy(target, rest string) *httputil.ReverseProxy { targetURL, err := url.Parse("http://" + target) if err != nil { diff --git a/versioned/internal/proxy/proxy_test.go b/versioned/internal/proxy/proxy_test.go index dbae426e73..18e47c3f82 100644 --- a/versioned/internal/proxy/proxy_test.go +++ b/versioned/internal/proxy/proxy_test.go @@ -111,8 +111,8 @@ func TestProxy_VersionlessObs_Primary(t *testing.T) { defer backend.Close() addr := strings.TrimPrefix(backend.URL, "http://") - // Lexicographic max is v2 — process-level obs pins to primary. - routes := newRoutes(map[string]string{"v1": "127.0.0.1:1", "v2": addr}) + // Numeric primary is v10 (not lexicographic "v2"). + routes := newRoutes(map[string]string{"v2": "127.0.0.1:1", "v10": addr}) handler := Handler(routes) srv := httptest.NewServer(handler) defer srv.Close() @@ -128,6 +128,30 @@ func TestProxy_VersionlessObs_Primary(t *testing.T) { } } +func TestProxy_VersionlessObs_PrimarySemverIsh(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "ok") + })) + defer backend.Close() + + addr := strings.TrimPrefix(backend.URL, "http://") + // Lexicographic max would be v0.2.9; numeric primary must be v0.2.11. + routes := newRoutes(map[string]string{"v0.2.9": "127.0.0.1:1", "v0.2.11": addr}) + handler := Handler(routes) + srv := httptest.NewServer(handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/stats/shards") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "ok" { + t.Fatalf("status=%d body=%q", resp.StatusCode, body) + } +} + func TestProxy_VersionlessObs_SessionFanout(t *testing.T) { miss := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { http.NotFound(w, nil) @@ -232,6 +256,7 @@ func TestProxy_VersionlessObs_LookupUnbound404(t *testing.T) { } func TestProxy_VersionlessObs_LookupErrorFallsBackToFanout(t *testing.T) { + resetLookupFanoutTelemetryForTest() hit := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "ok:%s", r.URL.Path) })) @@ -251,6 +276,19 @@ func TestProxy_VersionlessObs_LookupErrorFallsBackToFanout(t *testing.T) { if resp.StatusCode != http.StatusOK || string(body) != "ok:/sessions/42/diffs" { t.Fatalf("status=%d body=%q", resp.StatusCode, body) } + if got := LookupFanoutErrors(); got != 1 { + t.Fatalf("LookupFanoutErrors()=%d, want 1", got) + } + + // Second miss increments counter; warn is rate-limited (not asserted here). + resp2, err := http.Get(srv.URL + "/sessions/42/diffs") + if err != nil { + t.Fatal(err) + } + defer resp2.Body.Close() + if got := LookupFanoutErrors(); got != 2 { + t.Fatalf("LookupFanoutErrors()=%d, want 2", got) + } } func TestProxy_VersionlessObs_PayloadsStillVersioned(t *testing.T) { diff --git a/versioned/internal/proxy/version_sort.go b/versioned/internal/proxy/version_sort.go new file mode 100644 index 0000000000..7a7bf734b8 --- /dev/null +++ b/versioned/internal/proxy/version_sort.go @@ -0,0 +1,125 @@ +package proxy + +import ( + "sort" + "strconv" + "strings" +) + +// sortedVersions returns route map keys ordered from oldest to newest using +// compareVersionNames (numeric / dotted-numeric aware, not lexicographic). +func sortedVersions(routeMap map[string]string) []string { + versions := make([]string, 0, len(routeMap)) + for v := range routeMap { + versions = append(versions, v) + } + sort.SliceStable(versions, func(i, j int) bool { + return compareVersionNames(versions[i], versions[j]) < 0 + }) + return versions +} + +// primaryVersion picks the newest version name for process-level obs pinning. +func primaryVersion(versions []string) string { + if len(versions) == 0 { + return "" + } + best := versions[0] + for _, v := range versions[1:] { + if compareVersionNames(v, best) > 0 { + best = v + } + } + return best +} + +// compareVersionNames returns <0 if a is older than b, >0 if newer, 0 if equal. +// +// Names with a leading dotted-numeric prefix (optional leading "v"), such as +// "v2", "v10", "v0.2.11", compare by those numeric segments. That avoids +// lexicographic traps ("v10" < "v2", "v0.2.11" < "v0.2.9"). Non-numeric names +// (e.g. "dev") sort below any numeric name; among themselves they use +// strings.Compare. +func compareVersionNames(a, b string) int { + an, aok := versionNumericParts(a) + bn, bok := versionNumericParts(b) + switch { + case aok && bok: + if c := compareIntSlices(an, bn); c != 0 { + return c + } + return strings.Compare(a, b) + case aok && !bok: + return 1 + case !aok && bok: + return -1 + default: + return strings.Compare(a, b) + } +} + +func versionNumericParts(name string) ([]int, bool) { + s := strings.TrimSpace(name) + if s == "" { + return nil, false + } + if s[0] == 'v' || s[0] == 'V' { + s = s[1:] + } + if s == "" || s[0] < '0' || s[0] > '9' { + return nil, false + } + + var parts []int + i := 0 + for i < len(s) { + j := i + for j < len(s) && s[j] >= '0' && s[j] <= '9' { + j++ + } + if j == i { + break + } + n, err := strconv.Atoi(s[i:j]) + if err != nil { + return nil, false + } + parts = append(parts, n) + i = j + if i < len(s) && s[i] == '.' { + i++ + if i >= len(s) || s[i] < '0' || s[i] > '9' { + return nil, false // trailing or double dot + } + continue + } + break // suffix (e.g. -r2) or end + } + if len(parts) == 0 { + return nil, false + } + return parts, true +} + +func compareIntSlices(a, b []int) int { + n := len(a) + if len(b) < n { + n = len(b) + } + for i := 0; i < n; i++ { + if a[i] != b[i] { + if a[i] < b[i] { + return -1 + } + return 1 + } + } + switch { + case len(a) < len(b): + return -1 + case len(a) > len(b): + return 1 + default: + return 0 + } +} diff --git a/versioned/internal/proxy/version_sort_test.go b/versioned/internal/proxy/version_sort_test.go new file mode 100644 index 0000000000..2cff1a1320 --- /dev/null +++ b/versioned/internal/proxy/version_sort_test.go @@ -0,0 +1,70 @@ +package proxy + +import "testing" + +func TestCompareVersionNames(t *testing.T) { + tests := []struct { + a, b string + want int // sign of compareVersionNames(a,b) + }{ + {"v2", "v10", -1}, + {"v10", "v2", 1}, + {"v0.2.9", "v0.2.11", -1}, + {"v0.2.11", "v0.2.9", 1}, + {"v1", "v2", -1}, + {"v2", "v2", 0}, + {"2", "v10", -1}, + {"v0.2.11", "v0.2.11", 0}, + {"dev", "v2", -1}, // non-numeric sorts below numeric + {"v2", "dev", 1}, + {"alpha", "beta", -1}, + {"v0.2", "v0.2.0", -1}, // fewer segments sorts older when prefix equal + {"v0.2.0", "v0.2", 1}, + } + for _, tt := range tests { + got := compareVersionNames(tt.a, tt.b) + if sign(got) != tt.want { + t.Errorf("compareVersionNames(%q, %q) = %d, want sign %d", tt.a, tt.b, got, tt.want) + } + } +} + +func TestPrimaryVersion(t *testing.T) { + if got := primaryVersion([]string{"v2", "v10", "v1"}); got != "v10" { + t.Fatalf("primaryVersion(v1,v2,v10) = %q, want v10", got) + } + if got := primaryVersion([]string{"v0.2.9", "v0.2.11", "v0.2.10"}); got != "v0.2.11" { + t.Fatalf("primaryVersion(semver-ish) = %q, want v0.2.11", got) + } + if got := primaryVersion([]string{"v2", "dev"}); got != "v2" { + t.Fatalf("primaryVersion(v2,dev) = %q, want v2", got) + } +} + +func TestSortedVersions_NumericOrder(t *testing.T) { + got := sortedVersions(map[string]string{ + "v10": "a", + "v2": "b", + "v1": "c", + }) + want := []string{"v1", "v2", "v10"} + if len(got) != len(want) { + t.Fatalf("len=%d want %d: %v", len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("sortedVersions = %v, want %v", got, want) + } + } +} + +func sign(n int) int { + switch { + case n < 0: + return -1 + case n > 0: + return 1 + default: + return 0 + } +} From 8df65e06d786245bd34a551afd9856945bf4531d Mon Sep 17 00:00:00 2001 From: akup Date: Sat, 18 Jul 2026 11:20:24 +0300 Subject: [PATCH 3/5] fix(devshard): exclude settled sessions from stats detail lookup lookupBoundVersionActiveSession only checked meta version, so settled escrows could still resolve (and be revived). Match ListActiveSessions. --- devshard/cmd/devshardd/session/stats.go | 6 +++++- devshard/cmd/devshardd/session/stats_test.go | 22 ++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/devshard/cmd/devshardd/session/stats.go b/devshard/cmd/devshardd/session/stats.go index cd84713339..9d7d958474 100644 --- a/devshard/cmd/devshardd/session/stats.go +++ b/devshard/cmd/devshardd/session/stats.go @@ -230,18 +230,22 @@ func (m *HostManager) statsShardDetail(escrowID string, now time.Time) (*statsSh } // lookupBoundVersionActiveSession is O(1): one GetSessionMeta, no full-store scan. +// Only status=="active" sessions match (same filter as ListActiveSessions). // reason/metaVersion are set on not-found for negative-cache logging. func (m *HostManager) lookupBoundVersionActiveSession(escrowID string) (storage.ActiveSession, string, string, error) { meta, err := m.store.GetSessionMeta(escrowID) if err != nil { if errors.Is(err, storage.ErrSessionNotFound) { - return storage.ActiveSession{}, "absent_from_active_store", "", storage.ErrSessionNotFound + return storage.ActiveSession{}, "absent_from_store", "", storage.ErrSessionNotFound } return storage.ActiveSession{}, "meta_unreadable", "", err } if meta.Version != "" && meta.Version != m.boundVersion { return storage.ActiveSession{}, "version_mismatch", meta.Version, storage.ErrSessionNotFound } + if meta.Status != "active" { + return storage.ActiveSession{}, "not_active", meta.Version, storage.ErrSessionNotFound + } return storage.ActiveSession{EscrowID: escrowID, EpochID: meta.EpochID}, "", meta.Version, nil } diff --git a/devshard/cmd/devshardd/session/stats_test.go b/devshard/cmd/devshardd/session/stats_test.go index b305779b84..309f147efb 100644 --- a/devshard/cmd/devshardd/session/stats_test.go +++ b/devshard/cmd/devshardd/session/stats_test.go @@ -372,3 +372,25 @@ func TestStatsShardDetailNegativeCacheForVersionMismatch(t *testing.T) { require.Equal(t, http.StatusNotFound, rec2.Code) require.Equal(t, 1, counting.metaCalls, "version mismatch should be negatively cached") } + +func TestStatsShardDetailSkipsSettledSession(t *testing.T) { + base := newManagerTestStore(t) + _, _, hostSigner := createStoredSession(t, base, "escrow-settled", 7, 1) + require.NoError(t, base.MarkSettled("escrow-settled")) + + counting := &countingMetaStore{Storage: currentEpochStore{Storage: base, epoch: 7}} + mgr := NewHostManager(counting, hostSigner, stub.NewInferenceEngine(), stub.NewValidationEngine(), nil, testutil.RuntimeTestVersion, &mockBridge{}, nil, nil) + + rec1 := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/escrow-settled") + require.Equal(t, http.StatusNotFound, rec1.Code, "body: %s", rec1.Body.String()) + require.Equal(t, 1, counting.metaCalls) + require.Equal(t, 0, counting.listCalls) + + // Settled must not be revived into the live session map. + _, ok := mgr.existingServer("escrow-settled") + require.False(t, ok) + + rec2 := requestStats(t, mgr, statsTestRoutePrefix, "/stats/shards/escrow-settled") + require.Equal(t, http.StatusNotFound, rec2.Code) + require.Equal(t, 1, counting.metaCalls, "settled miss should be negatively cached") +} From a97e3bc8ff5b0f22e9f01332923d647fa8aec54f Mon Sep 17 00:00:00 2001 From: akup Date: Sat, 18 Jul 2026 15:45:40 +0300 Subject: [PATCH 4/5] Test hardening --- versioned/internal/proxy/proxy_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/versioned/internal/proxy/proxy_test.go b/versioned/internal/proxy/proxy_test.go index 35c674ffee..1717fb6f1c 100644 --- a/versioned/internal/proxy/proxy_test.go +++ b/versioned/internal/proxy/proxy_test.go @@ -398,6 +398,9 @@ func TestProxy_SSEStreaming(t *testing.T) { events = append(events, line) } } + if err := scanner.Err(); err != nil { + t.Fatalf("scan SSE body: %v", err) + } if len(events) != 3 { t.Errorf("got %d events, want 3", len(events)) } From 85b32fe51a911fe7d902309b5290dfbbe0b14ba1 Mon Sep 17 00:00:00 2001 From: akup Date: Sat, 18 Jul 2026 16:18:36 +0300 Subject: [PATCH 5/5] =?UTF-8?q?docs(devshard):=20add=20=C2=A74=20versionle?= =?UTF-8?q?ss=20observability=20test=20plan=20to=20v4=20release=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Give operators a walkthrough for unbound obs, legacy rewrite, owner-chat bind, PG routing, and rate limits, and surface the feature in the 0.2.14-v4 release guide. --- devshard/docs/release-0.2.14-v4.md | 51 ++++++- devshard/docs/v4-deploy-test-plan.md | 204 ++++++++++++++++++++++++++- 2 files changed, 247 insertions(+), 8 deletions(-) diff --git a/devshard/docs/release-0.2.14-v4.md b/devshard/docs/release-0.2.14-v4.md index b813149aa1..190f242c15 100644 --- a/devshard/docs/release-0.2.14-v4.md +++ b/devshard/docs/release-0.2.14-v4.md @@ -1,11 +1,12 @@ # Release guide: `devshard-0.2.14-v4` Operator-facing notes for the v4 line: multi-instance HA, Postgres storage, -gRPC-only gateway chain transport, and rollout constraints for mixed pre-v4 / -v4 estates. +gRPC-only gateway chain transport, versionless observability, and rollout +constraints for mixed pre-v4 / v4 estates. Detailed deploy verification: [v4-deploy-test-plan.md](./v4-deploy-test-plan.md). Architecture: [high-availability-architecture.md](./high-availability-architecture.md). +Observability: [pr-versionless-observability.md](./pr-versionless-observability.md). --- @@ -19,6 +20,11 @@ and validation-lease exclusivity. Pre-v4 approved versions stay **single-host** The gateway (`devshardctl`) talks to the chain over **gRPC only** — LCD REST create/settle paths are gone. +Public observability is **versionless**: dashboards scrape +`/devshard/sessions|stats|metrics` without binding protocol version. Only the +escrow owner binds via signed chat. Legacy `/devshard/{version}/…` obs URLs keep +working via join-proxy internal rewrite. + --- ## What's in this release @@ -31,6 +37,7 @@ create/settle paths are gone. | **Gateway transport** | Chain queries + escrow tx via gRPC; `--chain-rest` / `DEVSHARD_CHAIN_REST` removed | | **Settle confirm** | Settle waits for DeliverTx (`GetTx`) after SYNC CheckTx — same pattern as create | | **Failover** | Router retries another HA peer on first upstream 502 / connect failure | +| **Versionless obs** | Obs GETs never bind; owner chat binds; join rewrite + PG session lookup; `devshard_obs` rate limit | | **Status field** | Gateway `protocol_version` → `session_version` (bind / settlement tag) | | **testenv** | Docker citest S1–S9, G1–G4, A1–A4 (see [testenv/docs/scenarios.md](../testenv/docs/scenarios.md)) | @@ -102,6 +109,36 @@ literal `DEVSHARD_STORAGE_MODE=postgres` and `PGHOST` or the child returns See [storage-design.md](./storage-design.md) and [rolling-update.md](./rolling-update.md). +### Versionless observability (bind safety + canonical URLs) + +Observability GETs no longer call `CreateSession`. Unbound escrow obs returns +**404**; the first signed owner `POST …/chat/completions` binds the chosen +version. Prefer canonical monitor paths: + +| Preferred | Legacy (still works) | +| --- | --- | +| `/devshard/sessions/{id}/diffs` | `/devshard/{version}/sessions/{id}/diffs` (join proxy rewrite, no `Location`) | +| `/devshard/stats/shards/{id}` | `/devshard/{version}/stats/shards/{id}` | +| `/devshard/metrics` | `/devshard/{version}/metrics` | + +| Path | Meaning | +| --- | --- | +| `/devshard/healthz` | versiond supervisor | +| `/devshard/{version}/healthz` | that child (not rewritten) | + +With Postgres, versiond routes versionless session obs via `sessions.version` +(fan-out fallback if lookup disabled / SQLite). Join proxy rate-limits obs GETs +separately from chat: + +| Env | Where | Default | +| --- | --- | --- | +| `DEVSHARD_OBS_RATE_LIMIT_RPS` | join proxy | 10 | +| `DEVSHARD_OBS_BURST` | join proxy | 20 | +| `VERSIOND_DISABLE_SESSION_LOOKUP` | versiond | unset (lookup on when `PGHOST` set) | + +Manual walkthrough: [v4-deploy-test-plan.md](./v4-deploy-test-plan.md) §4. +Design note: [pr-versionless-observability.md](./pr-versionless-observability.md). + --- ## High-availability deployment @@ -176,6 +213,7 @@ Full checklists and negative proofs (multi-host + sqlite → 503, migrate invent | §1 Test deployment | NON_HA pin, sqlite→HA-fail→migrate→HA, mixed binding | | §2 Validation race | Same-key HA: one lease row per inference | | §3 High availability | Kill versiond → survivors serve (first-502); restart rejoins | +| §4 Versionless observability | Unbound obs 404; owner chat binds; rewrite; PG route; rate limit | --- @@ -188,8 +226,12 @@ Full checklists and negative proofs (multi-host + sqlite → 503, migrate invent - [ ] HA replicas share one `KEY_NAME` / keyring for the participant - [ ] Per-versiond SQLite volumes remain **distinct**; Postgres is shared - [ ] Monitors read `session_version`, not `protocol_version` +- [ ] Prefer versionless obs URLs (`/devshard/sessions|stats|metrics`); confirm + legacy `/devshard/{v}/…` still 200 with no public redirect +- [ ] Join proxy: set/tune `DEVSHARD_OBS_RATE_LIMIT_RPS` / `DEVSHARD_OBS_BURST` if needed - [ ] Smoke: chat/settle on a NON_HA path and on v4 HA path -- [ ] Optional: run §2 lease race and §3 kill/restart from the deploy test plan +- [ ] Optional: run §2 lease race, §3 kill/restart, and §4 versionless obs from + the deploy test plan --- @@ -211,7 +253,8 @@ with larger protocol bumps rather than a standalone migration. | Doc | Use | | --- | --- | -| [v4-deploy-test-plan.md](./v4-deploy-test-plan.md) | Deploy + three manual/operator test plans | +| [v4-deploy-test-plan.md](./v4-deploy-test-plan.md) | Deploy + four manual/operator test plans (§1–§4) | +| [pr-versionless-observability.md](./pr-versionless-observability.md) | Versionless obs design + unit/manual checklist | | [high-availability-architecture.md](./high-availability-architecture.md) | Current runtime topology | | [storage-design.md](./storage-design.md) | Storage mode selection | | [rolling-update.md](./rolling-update.md) | Binary swap / drain | diff --git a/devshard/docs/v4-deploy-test-plan.md b/devshard/docs/v4-deploy-test-plan.md index ecb7d093c3..ebe6feea83 100644 --- a/devshard/docs/v4-deploy-test-plan.md +++ b/devshard/docs/v4-deploy-test-plan.md @@ -1,7 +1,7 @@ # versiond high availability — Test plans This note covers **how to roll out** multi-instance HA + Postgres storage and -**three operator test plans**, including a routing constraint that the boot-migrate +**four operator test plans**, including a routing constraint that the boot-migrate path cannot paper over for already-deployed versions. | Plan | Section | What it proves | @@ -9,10 +9,12 @@ path cannot paper over for already-deployed versions. | **Test deployment plan** | §1 | Rollout phases, NON_HA pin, sqlite→migrate→HA, mixed binding | | **Validation race plan** | §2 | Same-key HA + Postgres: one validation lease per inference | | **High availability plan** | §3 | Kill versiond → survivors serve; restart → rejoins (check logs) | +| **Versionless observability plan** | §4 | Obs never binds version; rewrite + PG route; rate limit; health vs metrics | Related: [storage-design.md](./storage-design.md), [high-availability-architecture.md](./high-availability-architecture.md), -[release-0.2.14-v4.md](./release-0.2.14-v4.md). +[release-0.2.14-v4.md](./release-0.2.14-v4.md), +[pr-versionless-observability.md](./pr-versionless-observability.md). --- @@ -491,6 +493,8 @@ Checklist: loser does not double-submit — full walkthrough: **§2 Validation race plan** - [ ] Kill / restart HA versiond; survivors serve and restarted host rejoins — **§3 High availability plan** (verify in logs) +- [ ] Versionless obs: unbound 404, owner chat binds, legacy rewrite — + **§4 Versionless observability plan** ### 1.10 Follow-up test (out of scope until tool exists) @@ -1031,7 +1035,197 @@ make down --- -## 4. Summary for operators +## 4. Versionless observability plan (manual) + +Companion to [pr-versionless-observability.md](./pr-versionless-observability.md). +Unit coverage already exists (`versioned/internal/proxy`, `SessionServerExisting`, +owner-chat bind). This section is the **operator walkthrough**. + +**Goal:** public observability never binds protocol version; only the escrow +owner binds via signed chat; versionless + legacy rewrite work; Postgres lookup +routes session obs to the bound child; obs GETs are rate-limited separately from +chat. + +```text +Dashboard GET /devshard/v2/sessions/E/diffs + → join proxy rewrite (no Location) → /devshard/sessions/E/diffs + → versiond (PG lookup or fan-out) → child + → SessionServerExisting (no CreateSession) + → 404 if unbound + +Creator POST /devshard/v3/sessions/E/chat/completions (owner sig) + → BindOwnerChat → CreateSession(Version=v3) +``` + +**Out of scope:** multi-version merge of `/stats/shards` list; migrating already +wrongly bound escrows; lease race (§2); kill/restart (§3). + +### 4.0 Preconditions + +| Requirement | Why | +| --- | --- | +| ≥1 approved HA version (testenv: `v2`) | Bind + obs targets | +| Shared Postgres (`PGHOST`) on versiond | Bound-version lookup for versionless session obs | +| Gateway for owner chat | Only signed chat binds | +| **Join / genesis proxy** for rewrite + rate limit | Bare versiond-router has no `devshard_obs` rewrite zone | + +**Stacks:** + +| Cases | Stack | +| --- | --- | +| §4.1, §4.3, §4.4, §4.5 | testenv multi + Postgres (router `:8080`, gateway `:8081`) — use `/v2/…` or `/sessions/…` on the router; versionless paths work without join rewrite | +| §4.2 rewrite, §4.6 rate limit | **Join / local-test-net with public proxy** in front of versiond (`/devshard/…`) | + +Env (see also [proxy/README.md](../../proxy/README.md)): + +| Variable | Where | Role | +| --- | --- | --- | +| `PGHOST` / `DATABASE_URL` | versiond | Enable session-version lookup | +| `VERSIOND_DISABLE_SESSION_LOOKUP` | versiond | Force fan-out even when PG is set | +| `DEVSHARD_OBS_RATE_LIMIT_RPS` | join proxy | Per-IP obs GET limit (default 10) | +| `DEVSHARD_OBS_BURST` | join proxy | Obs burst (default 20) | + +### 4.1 Unbound obs does not bind + +Pick an escrow id that has **no** session yet (new chain escrow, or a fake id +that will 404). + +```bash +# testenv router (no /devshard/ prefix) +curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1:8080/sessions/UNBOUND-ESCROW/diffs" +curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1:8080/v2/sessions/UNBOUND-ESCROW/diffs" + +# join / public proxy +curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1/devshard/sessions/UNBOUND-ESCROW/diffs" +curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1/devshard/v2/sessions/UNBOUND-ESCROW/diffs" +``` + +| Check | Pass | +| --- | --- | +| HTTP status | **404** (not 200) | +| Side effect | No new row in `devshard_sessions` / session index for that escrow; no accidental version stamp | + +### 4.2 Legacy rewrite (join proxy) + +Requires the **public proxy** rewrite path (not bare versiond-router). + +1. Bind an escrow via owner chat on a chosen version (e.g. `v2`) — §4.3. +2. Hit the **legacy** versioned obs URL and the canonical versionless URL: + +```bash +# After bind; replace ESCROW and adjust host/port for join proxy +curl -sSI "http://127.0.0.1/devshard/v2/sessions/${ESCROW}/diffs" \ + | grep -Ei 'HTTP/|^[Ll]ocation:' +curl -sS "http://127.0.0.1/devshard/v2/sessions/${ESCROW}/diffs" -o /tmp/legacy-diffs.body +curl -sS "http://127.0.0.1/devshard/sessions/${ESCROW}/diffs" -o /tmp/canon-diffs.body +cmp /tmp/legacy-diffs.body /tmp/canon-diffs.body && echo bodies_match +``` + +| Check | Pass | Fail | +| --- | --- | --- | +| Status | **200** (bound escrow) | 308/301 to another path, or 404 after bind | +| `Location` | **Absent** (internal `rewrite … last`, not public redirect) | Client-visible redirect | +| Body | Legacy and versionless responses match | Different payloads | + +### 4.3 Owner chat binds chosen version + +1. Create/bind escrow through the **gateway** (owner key). +2. Confirm obs still **404** before first chat (§4.1). +3. Owner chat on the intended version path (gateway OpenAI path, or signed + `POST /…/v{ver}/sessions/{id}/chat/completions`). +4. Re-check obs: + +```bash +# testenv router examples after gateway chat bound escrow ESCROW to v2 +curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1:8080/sessions/${ESCROW}/diffs" +curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1:8080/stats/shards/${ESCROW}" +``` + +| Check | Pass | +| --- | --- | +| Before chat | Obs **404** | +| After owner chat | Diffs + stats detail **200** | +| Bind source | Only owner chat created the session (not a prior obs GET) | + +### 4.4 PG routing (HA + lookup) + +With `PGHOST` set and session bound (Postgres holds `sessions.version`): + +1. Drive versionless `GET …/sessions/${ESCROW}/diffs` (via router or proxy). +2. Confirm traffic hits the **bound** version’s child (versiond logs / child + access logs; with multi-versiond sticky router, correlate escrow → upstream). +3. Optional negative: set `VERSIOND_DISABLE_SESSION_LOOKUP=true`, recreate + versiond — versionless obs still succeeds via **fan-out** (may be slower; + lookup-error / fan-out warn may appear when PG errors). + +| Check | Pass | +| --- | --- | +| Default PG lookup | Bound child serves diffs; unbound still 404 | +| Lookup disabled | Fan-out still finds bound session when a child has it | + +### 4.5 Health vs metrics + +```bash +# Join / public proxy shapes (adjust host). testenv router: omit /devshard/ +curl -sS -o /dev/null -w "%{http_code}\n" "http://127.0.0.1/devshard/healthz" +curl -sS -o /dev/null -w "%{http_code}\n" "http://127.0.0.1/devshard/v2/healthz" +curl -sS -o /dev/null -w "%{http_code}\n" "http://127.0.0.1/devshard/metrics" +``` + +| Path | Expect | +| --- | --- | +| `/devshard/healthz` (or router `/healthz`) | **versiond supervisor** status — not a child process metrics scrape | +| `/devshard/{version}/healthz` | **That child** health (not rewritten to versionless) | +| `/devshard/metrics` | Pins **newest** child by numeric/dotted version sort (`v10` > `v2`) | + +### 4.6 Obs rate limit (join proxy) + +Requires public proxy `devshard_obs` zone. Temporarily set e.g. +`DEVSHARD_OBS_RATE_LIMIT_RPS=2`, `DEVSHARD_OBS_BURST=2`, reload proxy. + +```bash +# Burst obs GETs — expect some 503 +for i in $(seq 1 30); do + curl -sS -o /dev/null -w "%{http_code}\n" \ + "http://127.0.0.1/devshard/metrics" +done | sort | uniq -c + +# Chat must remain exempt (gateway or signed chat POST still succeeds) +``` + +| Check | Pass | Fail | +| --- | --- | --- | +| Obs burst | Some responses **503** under low RPS | All 200 under deliberate flood | +| Chat | Owner chat / gateway inference still **2xx** | Chat also rate-limited by `devshard_obs` | + +Restore default obs rate env after the exercise. + +### 4.7 Checklist summary + +- [ ] Unbound obs → **404**; no session bind (§4.1) +- [ ] Legacy `/devshard/{v}/sessions/…/diffs` → **200**, no `Location` (§4.2) +- [ ] Owner chat binds; then versionless diffs/stats **200** (§4.3) +- [ ] PG lookup routes to bound child; fan-out works if lookup disabled (§4.4) +- [ ] Supervisor `/healthz` vs `/{v}/healthz` vs `/metrics` (§4.5) +- [ ] Obs rate limit 503; chat exempt (§4.6) + +### 4.8 Cleanup + +```bash +cd devshard/testenv +make down +# Restore DEVSHARD_OBS_* / VERSIOND_DISABLE_SESSION_LOOKUP on join stacks +``` + +--- + +## 5. Summary for operators 1. **HA multi-instance + shared Postgres applies to versions outside `VERSIOND_NON_HA_VERSIONS`**, not to already-deployed pre-HA binaries. @@ -1049,7 +1243,9 @@ make down versions, plus the sqlite→HA-fail→migrate sequence — not “all migrate to Postgres” blindly. -7. **Three test plans in this doc:** +7. **Four test plans in this doc:** - **§1 Test deployment plan** — rollout, routing, migrate, mixed binding. - **§2 Validation race plan** — lease exclusivity under same-key HA. - **§3 High availability plan** — kill / restart versiond; verify via logs. + - **§4 Versionless observability plan** — obs never binds; rewrite; PG route; + rate limit; health vs metrics.