From e47bf4e082d6a11127a098bf6598a9a800f91088 Mon Sep 17 00:00:00 2001 From: Chai Bot Date: Wed, 19 Aug 2026 13:37:34 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Return=20503=20or=20404=20for=20?= =?UTF-8?q?unavailable=20catalog=20content=20based=20on=20leadership?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a catalog's content is not found (fs.ErrNotExist), the HTTP response now depends on whether the current pod is the leader: - Leader: returns 404 Not Found, because the leader has reconciled all catalogs and the content genuinely does not exist. - Non-leader: returns 503 Service Unavailable with Retry-After: 1, because the content may exist on the leader but has not been synced to this replica yet. Previously, all pods unconditionally returned 503 for missing content, which told clients to retry forever even when the leader knew the catalog did not exist. The leadership status is derived from the controller-runtime manager's Elected() channel, which closes when the pod wins the leader lease (or immediately when leader election is disabled). Co-Authored-By: Claude Opus 4.6 --- cmd/catalogd/main.go | 19 +++- internal/catalogd/server/handlers.go | 57 ++++++---- internal/catalogd/server/handlers_test.go | 102 ++++++++++++++++-- .../server/http_preconditions_check.go | 14 ++- internal/catalogd/serverutil/serverutil.go | 8 +- internal/catalogd/storage/localdir.go | 11 +- internal/catalogd/storage/localdir_test.go | 50 ++++++++- 7 files changed, 221 insertions(+), 40 deletions(-) diff --git a/cmd/catalogd/main.go b/cmd/catalogd/main.go index 2b6f81b4ab..58bfc2486c 100644 --- a/cmd/catalogd/main.go +++ b/cmd/catalogd/main.go @@ -350,7 +350,7 @@ func run(ctx context.Context) error { }, } - var localStorage storage.Instance + var localStorage *storage.LocalDirV1 metrics.Registry.MustRegister(catalogdmetrics.RequestDurationMetric) storeDir := filepath.Join(cfg.cacheDir, storageDir) @@ -386,6 +386,23 @@ func run(ctx context.Context) error { graphqlMode, ) + // Leadership detection: mgr.Elected() is a channel that closes once this pod + // wins the leader lease. When leader election is disabled (standalone mode), + // we leave IsLeader nil so the handler defaults to leader behavior (404 for + // missing catalogs). This avoids a startup window where mgr.Elected() is not + // yet closed and the handler would incorrectly return 503. + if cfg.enableLeaderElection { + elected := mgr.Elected() + localStorage.IsLeader = func() bool { + select { + case <-elected: + return true + default: + return false + } + } + } + // Config for the catalogd web server catalogServerConfig := serverutil.CatalogServerConfig{ ExternalAddr: cfg.externalAddr, diff --git a/internal/catalogd/server/handlers.go b/internal/catalogd/server/handlers.go index 7c90dc1147..410d15294b 100644 --- a/internal/catalogd/server/handlers.go +++ b/internal/catalogd/server/handlers.go @@ -53,6 +53,7 @@ type CatalogHandlers struct { rootURL *url.URL enableMetas MetasHandlerMode enableGraphQL GraphQLQueriesMode + isLeader func() bool } // Index provides methods for looking up catalog content by schema/package/name @@ -72,14 +73,19 @@ type CatalogStore interface { GetIndex(catalog string) (Index, error) } -// NewCatalogHandlers creates a new HTTP handlers instance -func NewCatalogHandlers(store CatalogStore, graphqlSvc service.GraphQLService, rootURL *url.URL, enableMetas MetasHandlerMode, enableGraphQL GraphQLQueriesMode) *CatalogHandlers { +// NewCatalogHandlers creates a new HTTP handlers instance. +// isLeader reports whether the current pod is the leader-elected instance. +// When a catalog's content is not found (fs.ErrNotExist), the leader returns +// 404 (the catalog genuinely does not exist) while a non-leader returns 503 +// with Retry-After (content may not have been synced to this replica yet). +func NewCatalogHandlers(store CatalogStore, graphqlSvc service.GraphQLService, rootURL *url.URL, enableMetas MetasHandlerMode, enableGraphQL GraphQLQueriesMode, isLeader func() bool) *CatalogHandlers { return &CatalogHandlers{ store: store, graphqlSvc: graphqlSvc, rootURL: rootURL, enableMetas: enableMetas, enableGraphQL: enableGraphQL, + isLeader: isLeader, } } @@ -117,12 +123,12 @@ func (h *CatalogHandlers) Handler() http.Handler { func (h *CatalogHandlers) handleV1All(w http.ResponseWriter, r *http.Request) { catalog := r.PathValue("catalog") if err := isValidCatalogName(catalog); err != nil { - httpError(w, err) + h.httpError(w, err) return } catalogFile, catalogStat, err := h.store.GetCatalogData(catalog) if err != nil { - httpError(w, err) + h.httpError(w, err) return } defer catalogFile.Close() @@ -135,7 +141,7 @@ func (h *CatalogHandlers) handleV1All(w http.ResponseWriter, r *http.Request) { func (h *CatalogHandlers) handleV1Metas(w http.ResponseWriter, r *http.Request) { catalog := r.PathValue("catalog") if err := isValidCatalogName(catalog); err != nil { - httpError(w, err) + h.httpError(w, err) return } @@ -148,13 +154,13 @@ func (h *CatalogHandlers) handleV1Metas(w http.ResponseWriter, r *http.Request) for param := range r.URL.Query() { if !expectedParams[param] { - httpError(w, errInvalidParams) + h.httpError(w, errInvalidParams) return } } catalogFile, catalogStat, err := h.store.GetCatalogData(catalog) if err != nil { - httpError(w, err) + h.httpError(w, err) return } defer catalogFile.Close() @@ -171,17 +177,17 @@ func (h *CatalogHandlers) handleV1Metas(w http.ResponseWriter, r *http.Request) if schema == "" && pkg == "" && name == "" { // If no parameters are provided, return the entire catalog - serveJSONLines(w, r, catalogFile) + h.serveJSONLines(w, r, catalogFile) return } idx, err := h.store.GetIndex(catalog) if err != nil { - httpError(w, err) + h.httpError(w, err) return } indexReader := idx.Get(catalogFile, schema, pkg, name) - serveJSONLines(w, r, indexReader) + h.serveJSONLines(w, r, indexReader) } // handleV1GraphQL handles GraphQL queries @@ -197,7 +203,7 @@ func (h *CatalogHandlers) handleV1GraphQL(w http.ResponseWriter, r *http.Request catalog := r.PathValue("catalog") if err := isValidCatalogName(catalog); err != nil { - httpError(w, err) + h.httpError(w, err) return } @@ -225,25 +231,38 @@ func (h *CatalogHandlers) handleV1GraphQL(w http.ResponseWriter, r *http.Request result, err := h.graphqlSvc.ExecuteQuery(r.Context(), catalog, params.Query) if err != nil { - httpError(w, err) + h.httpError(w, err) return } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(result); err != nil { - httpError(w, err) + h.httpError(w, err) return } } -// httpError writes an HTTP error response based on the error type -func httpError(w http.ResponseWriter, err error) { +// httpError writes an HTTP error response based on the error type. +// For fs.ErrNotExist, the response depends on leadership status: +// - Leader: 404 Not Found (the catalog genuinely does not exist) +// - Non-leader: 503 Service Unavailable + Retry-After: 1 (content may exist on the leader) +func (h *CatalogHandlers) httpError(w http.ResponseWriter, err error) { var code int var message string switch { case errors.Is(err, fs.ErrNotExist): - code = http.StatusNotFound - message = fmt.Sprintf("%d %s", code, http.StatusText(code)) + if h.isLeader() { + // Leader has reconciled all catalogs; if content is still not + // found, the catalog genuinely does not exist. + code = http.StatusNotFound + message = fmt.Sprintf("%d %s", code, http.StatusText(code)) + } else { + // Non-leader pods may not have synced content yet; tell the + // client to retry (potentially against the leader). + code = http.StatusServiceUnavailable + w.Header().Set("Retry-After", "1") + message = "catalog content not yet available" + } case errors.Is(err, fs.ErrPermission): code = http.StatusForbidden message = fmt.Sprintf("%d %s", code, http.StatusText(code)) @@ -268,7 +287,7 @@ func httpError(w http.ResponseWriter, err error) { } // serveJSONLines writes JSON lines content to the response -func serveJSONLines(w http.ResponseWriter, r *http.Request, rs io.Reader) { +func (h *CatalogHandlers) serveJSONLines(w http.ResponseWriter, r *http.Request, rs io.Reader) { w.Header().Add("Content-Type", "application/jsonl") // Copy the content of the reader to the response writer only if it's a GET request if r.Method == http.MethodHead { @@ -276,7 +295,7 @@ func serveJSONLines(w http.ResponseWriter, r *http.Request, rs io.Reader) { } _, err := io.Copy(w, rs) if err != nil { - httpError(w, err) + h.httpError(w, err) return } } diff --git a/internal/catalogd/server/handlers_test.go b/internal/catalogd/server/handlers_test.go index 3b5ec604c1..3224516817 100644 --- a/internal/catalogd/server/handlers_test.go +++ b/internal/catalogd/server/handlers_test.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "os" "strings" "testing" @@ -19,13 +20,15 @@ import ( mockcatalogdservice "github.com/operator-framework/operator-controller/internal/testutil/mock/catalogdservice" ) +var alwaysLeader = func() bool { return true } + func TestHandleV1GraphQL_MethodNotAllowed(t *testing.T) { ctrl := gomock.NewController(t) rootURL, _ := url.Parse("http://localhost/") store := mockcatalogdserver.NewMockCatalogStore(ctrl) graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() req := httptest.NewRequest(http.MethodGet, "/test-catalog/api/v1/graphql", nil) @@ -44,7 +47,7 @@ func TestHandleV1GraphQL_InvalidCatalogName(t *testing.T) { store := mockcatalogdserver.NewMockCatalogStore(ctrl) graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() req := httptest.NewRequest(http.MethodPost, "/INVALID-CATALOG-NAME/api/v1/graphql", strings.NewReader(`{"query": "{ summary { totalSchemas } }"}`)) @@ -63,7 +66,7 @@ func TestHandleV1GraphQL_InvalidJSON(t *testing.T) { store := mockcatalogdserver.NewMockCatalogStore(ctrl) graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() req := httptest.NewRequest(http.MethodPost, "/test-catalog/api/v1/graphql", strings.NewReader(`{invalid json`)) @@ -82,7 +85,7 @@ func TestHandleV1GraphQL_EmptyQuery(t *testing.T) { store := mockcatalogdserver.NewMockCatalogStore(ctrl) graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() req := httptest.NewRequest(http.MethodPost, "/test-catalog/api/v1/graphql", strings.NewReader(`{"query": ""}`)) @@ -104,7 +107,7 @@ func TestHandleV1GraphQL_QueryTooLarge(t *testing.T) { store := mockcatalogdserver.NewMockCatalogStore(ctrl) graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() // Create a query larger than 100KB @@ -125,7 +128,7 @@ func TestHandleV1GraphQL_BodyTooLarge(t *testing.T) { store := mockcatalogdserver.NewMockCatalogStore(ctrl) graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() // Create a body larger than 1MB @@ -158,7 +161,7 @@ func TestHandleV1GraphQL_Success(t *testing.T) { graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) graphqlSvc.EXPECT().ExecuteQuery(gomock.Any(), "test-catalog", "{ summary { totalSchemas } }").Return(expectedResult, nil) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() query := `{"query": "{ summary { totalSchemas } }"}` @@ -195,7 +198,7 @@ func TestHandleV1GraphQL_Success(t *testing.T) { } } -func TestHandleV1GraphQL_CatalogNotFoundError(t *testing.T) { +func TestHandleV1GraphQL_CatalogNotFoundError_Leader(t *testing.T) { ctrl := gomock.NewController(t) rootURL, _ := url.Parse("http://localhost/") @@ -204,7 +207,7 @@ func TestHandleV1GraphQL_CatalogNotFoundError(t *testing.T) { graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) graphqlSvc.EXPECT().ExecuteQuery(gomock.Any(), "test-catalog", "{ summary { totalSchemas } }").Return(nil, fs.ErrNotExist) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() query := `{"query": "{ summary { totalSchemas } }"}` @@ -213,11 +216,43 @@ func TestHandleV1GraphQL_CatalogNotFoundError(t *testing.T) { handler.ServeHTTP(w, req) + // Leader knows the catalog genuinely does not exist → 404 if w.Code != http.StatusNotFound { t.Errorf("Expected status %d, got %d", http.StatusNotFound, w.Code) } } +func TestHandleV1GraphQL_CatalogNotFoundError_NonLeader(t *testing.T) { + ctrl := gomock.NewController(t) + rootURL, _ := url.Parse("http://localhost/") + + store := mockcatalogdserver.NewMockCatalogStore(ctrl) + + graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) + graphqlSvc.EXPECT().ExecuteQuery(gomock.Any(), "test-catalog", "{ summary { totalSchemas } }").Return(nil, fs.ErrNotExist) + + neverLeader := func() bool { return false } + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, neverLeader) + handler := handlers.Handler() + + query := `{"query": "{ summary { totalSchemas } }"}` + req := httptest.NewRequest(http.MethodPost, "/test-catalog/api/v1/graphql", strings.NewReader(query)) + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + // Non-leader may not have synced content yet → 503 with Retry-After + if w.Code != http.StatusServiceUnavailable { + t.Errorf("Expected status %d, got %d", http.StatusServiceUnavailable, w.Code) + } + if retryAfter := w.Header().Get("Retry-After"); retryAfter != "1" { + t.Errorf("Expected Retry-After header '1', got '%s'", retryAfter) + } + if body := strings.TrimSpace(w.Body.String()); body != "catalog content not yet available" { + t.Errorf("Expected body 'catalog content not yet available', got '%s'", body) + } +} + func TestHandleV1GraphQL_ExecuteQueryError(t *testing.T) { ctrl := gomock.NewController(t) rootURL, _ := url.Parse("http://localhost/") @@ -227,7 +262,7 @@ func TestHandleV1GraphQL_ExecuteQueryError(t *testing.T) { graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) graphqlSvc.EXPECT().ExecuteQuery(gomock.Any(), "test-catalog", "{ summary { totalSchemas } }").Return(nil, context.DeadlineExceeded) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() query := `{"query": "{ summary { totalSchemas } }"}` @@ -249,7 +284,7 @@ func TestAllowedMethodsHandler_POSTOnlyForGraphQL(t *testing.T) { graphqlSvc := mockcatalogdservice.NewMockGraphQLService(ctrl) graphqlSvc.EXPECT().ExecuteQuery(gomock.Any(), "test-catalog", "{ summary { totalSchemas } }").Return(nil, nil) - handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled) + handlers := server.NewCatalogHandlers(store, graphqlSvc, rootURL, server.MetasHandlerDisabled, server.GraphQLQueriesEnabled, alwaysLeader) handler := handlers.Handler() // Test POST to GraphQL endpoint - should be allowed @@ -263,3 +298,48 @@ func TestAllowedMethodsHandler_POSTOnlyForGraphQL(t *testing.T) { t.Error("POST should be allowed for GraphQL endpoint at router level") } } + +func TestHandleV1Metas_InvalidIfModifiedSince(t *testing.T) { + ctrl := gomock.NewController(t) + rootURL, _ := url.Parse("http://localhost/") + + // Create a temporary file with catalog content to be returned by the mock. + tmpFile, err := os.CreateTemp(t.TempDir(), "catalog-*.jsonl") + if err != nil { + t.Fatal(err) + } + catalogData := `{"schema":"olm.package","name":"test-pkg"}` + "\n" + if _, err := tmpFile.WriteString(catalogData); err != nil { + t.Fatal(err) + } + // Seek back to the beginning so the handler can read it. + if _, err := tmpFile.Seek(0, 0); err != nil { + t.Fatal(err) + } + info, err := tmpFile.Stat() + if err != nil { + t.Fatal(err) + } + + store := mockcatalogdserver.NewMockCatalogStore(ctrl) + store.EXPECT().GetCatalogData("test-catalog").Return(tmpFile, info, nil) + + handlers := server.NewCatalogHandlers(store, nil, rootURL, server.MetasHandlerEnabled, server.GraphQLQueriesDisabled, alwaysLeader) + handler := handlers.Handler() + + req := httptest.NewRequest(http.MethodGet, "/test-catalog/api/v1/metas", nil) + req.Header.Set("If-Modified-Since", "this-is-not-a-valid-date") + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + // An invalid If-Modified-Since header must produce only a 500 error; + // no catalog JSONL should be appended to the body. + if w.Code != http.StatusInternalServerError { + t.Errorf("Expected status %d, got %d", http.StatusInternalServerError, w.Code) + } + body := w.Body.String() + if strings.Contains(body, "olm.package") { + t.Errorf("Response body must not contain catalog JSONL data, got: %s", body) + } +} diff --git a/internal/catalogd/server/http_preconditions_check.go b/internal/catalogd/server/http_preconditions_check.go index ceabad2d30..ea267a8275 100644 --- a/internal/catalogd/server/http_preconditions_check.go +++ b/internal/catalogd/server/http_preconditions_check.go @@ -8,10 +8,13 @@ package server import ( + "fmt" "net/http" "net/textproto" "strings" "time" + + "k8s.io/klog/v2" ) type condResult int @@ -20,6 +23,7 @@ const ( condNone condResult = iota condTrue condFalse + condError // an error response has already been written ) // checkPreconditions evaluates request preconditions and reports whether a precondition @@ -44,9 +48,12 @@ func checkPreconditions(w http.ResponseWriter, r *http.Request, modtime time.Tim return true } case condNone: - if checkIfModifiedSince(r, w, modtime) == condFalse { + switch checkIfModifiedSince(r, w, modtime) { + case condFalse: writeNotModified(w) return true + case condError: + return true } } return false @@ -59,8 +66,9 @@ func checkIfModifiedSince(r *http.Request, w http.ResponseWriter, modtime time.T } t, err := parseTime(ims) if err != nil { - httpError(w, err) - return condNone + klog.ErrorS(err, "HTTP error parsing If-Modified-Since header", "code", http.StatusInternalServerError) + http.Error(w, fmt.Sprintf("%d %s", http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)), http.StatusInternalServerError) + return condError } // The Last-Modified header truncates sub-second precision so // the modtime needs to be truncated too. diff --git a/internal/catalogd/serverutil/serverutil.go b/internal/catalogd/serverutil/serverutil.go index 3a597de877..81104a87bc 100644 --- a/internal/catalogd/serverutil/serverutil.go +++ b/internal/catalogd/serverutil/serverutil.go @@ -36,7 +36,7 @@ type CatalogServerConfig struct { // a readiness check that passes once the server has started serving. Because // NeedLeaderElection returns false, Start() is called on every pod immediately, so all // replicas bind the catalog port and become ready. Non-leader pods serve requests but -// return 404 (empty local cache); callers are expected to retry. +// return 503 Service Unavailable (empty local cache); callers are expected to retry. func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig) error { shutdownTimeout := 30 * time.Second r := &catalogServerRunnable{ @@ -57,7 +57,7 @@ func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig) error // Register a readiness check that passes once Start() has been called and the // server is actively serving. All pods reach Start() (NeedLeaderElection=false), - // so all replicas become ready and receive traffic; non-leaders return 404 until + // so all replicas become ready and receive traffic; non-leaders return 503 until // they win the leader lease and populate their local cache. if err := mgr.AddReadyzCheck("catalog-server", r.readyzCheck()); err != nil { return fmt.Errorf("error adding catalog server readiness check: %w", err) @@ -68,7 +68,7 @@ func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig) error // catalogServerRunnable is a Runnable that binds the catalog HTTP port on every pod. // Because NeedLeaderElection returns false, Start() is called on all replicas immediately; -// non-leader pods serve the catalog port but return 404 (empty local cache). +// non-leader pods serve the catalog port but return 503 Service Unavailable (empty local cache). type catalogServerRunnable struct { cfg CatalogServerConfig server *http.Server @@ -85,7 +85,7 @@ type catalogServerRunnable struct { // // Non-leader pods serve the catalog HTTP port but have an empty local cache // (only the leader's reconciler downloads catalog content), so requests to a -// non-leader return 404. Callers are expected to retry. +// non-leader return 503 Service Unavailable. Callers are expected to retry. func (r *catalogServerRunnable) NeedLeaderElection() bool { return false } func (r *catalogServerRunnable) Start(ctx context.Context) error { diff --git a/internal/catalogd/storage/localdir.go b/internal/catalogd/storage/localdir.go index 0cd65933b0..86b1444630 100644 --- a/internal/catalogd/storage/localdir.go +++ b/internal/catalogd/storage/localdir.go @@ -48,6 +48,10 @@ type LocalDirV1 struct { EnableMetasHandler MetasHandlerMode EnableGraphQLQueries GraphQLQueriesMode + // IsLeader reports whether the current pod is the leader-elected instance. + // When nil, the pod is assumed to be the leader (single-replica / standalone). + IsLeader func() bool + m sync.RWMutex // this singleflight Group is used in `GetIndex()` to handle concurrent HTTP requests // optimally. With the use of this singleflight group, the index is loaded from disk @@ -413,7 +417,12 @@ func (s *LocalDirV1) BaseURL(catalog string) string { // StorageServerHandler returns an HTTP handler for serving catalog content // This implements the Instance interface for backward compatibility func (s *LocalDirV1) StorageServerHandler() http.Handler { - handlers := server.NewCatalogHandlers(s, s.graphqlSvc, s.RootURL, s.EnableMetasHandler, s.EnableGraphQLQueries) + isLeader := s.IsLeader + if isLeader == nil { + // Default: assume leader (single-replica / standalone mode). + isLeader = func() bool { return true } + } + handlers := server.NewCatalogHandlers(s, s.graphqlSvc, s.RootURL, s.EnableMetasHandler, s.EnableGraphQLQueries, isLeader) return handlers.Handler() } diff --git a/internal/catalogd/storage/localdir_test.go b/internal/catalogd/storage/localdir_test.go index 76efc00e7f..5e21256d81 100644 --- a/internal/catalogd/storage/localdir_test.go +++ b/internal/catalogd/storage/localdir_test.go @@ -306,7 +306,7 @@ func TestLocalDirServerHandler(t *testing.T) { URLPath: "/catalogs/test-catalog.jsonl", }, { - name: "Server returns 404 when non-existent catalog is queried", + name: "Server returns 404 when non-existent catalog is queried (leader)", expectedStatusCode: http.StatusNotFound, expectedContent: "404 Not Found", URLPath: "/catalogs/non-existent-catalog/api/v1/all", @@ -341,6 +341,54 @@ func TestLocalDirServerHandler(t *testing.T) { } } +func TestLocalDirServerHandler_NonLeader(t *testing.T) { + store := NewLocalDirV1(t.TempDir(), &url.URL{Path: urlPrefix}, MetasHandlerDisabled, GraphQLQueriesDisabled) + // Mark this pod as a non-leader so that missing content returns 503. + store.IsLeader = func() bool { return false } + // Do NOT store any catalog data — the non-leader has an empty cache. + + testServer := httptest.NewServer(store.StorageServerHandler()) + defer testServer.Close() + + req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/catalogs/some-catalog/api/v1/all", testServer.URL), nil) + require.NoError(t, err) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + + // Non-leader should return 503 with Retry-After instead of 404. + require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + assert.Equal(t, "1", resp.Header.Get("Retry-After")) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, "catalog content not yet available", strings.TrimSpace(string(body))) + require.NoError(t, resp.Body.Close()) +} + +func TestLocalDirServerHandler_StandaloneMode(t *testing.T) { + // Standalone mode: IsLeader is nil (leader election disabled). + // The handler should default to leader behavior and return 404 + // for nonexistent catalogs, not 503. + store := NewLocalDirV1(t.TempDir(), &url.URL{Path: urlPrefix}, MetasHandlerDisabled, GraphQLQueriesDisabled) + // Do NOT set store.IsLeader — leave it nil to simulate standalone mode. + + testServer := httptest.NewServer(store.StorageServerHandler()) + defer testServer.Close() + + req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/catalogs/non-existent-catalog/api/v1/all", testServer.URL), nil) + require.NoError(t, err) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + + // Standalone (no leader election) should return 404 for nonexistent catalogs. + require.Equal(t, http.StatusNotFound, resp.StatusCode) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, "404 Not Found", strings.TrimSpace(string(body))) + require.NoError(t, resp.Body.Close()) +} + // Tests to verify the behavior of the metas endpoint, as described in // https://docs.google.com/document/d/1s6_9IFEKGQLNh3ueH7SF4Yrx4PW9NSiNFqFIJx0pU-8/ func TestMetasEndpoint(t *testing.T) {