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) {