Fix underfilled pages in ListSources and ListRegistries (#688)

## Summary
- `ListSources` fetched a single batch of up to `MaxPageSize` records
and then
post-filtered by caller JWT claims, causing the returned list to be
underfilled
when filtering removed rows and more qualifying records existed in the
DB.
- `ListRegistries` fetched all registries in a single query and
post-filtered by
    claims, with the same result.             
Both functions now use a streaming batch loop — identical in structure
to the
existing `streamSkillRows` — that keeps fetching until the DB is
exhausted,
ensuring all qualifying records are always returned.
`ListRegistries` required a small SQL change to add cursor-based
pagination
(`WHERE name > cursor ORDER BY name LIMIT size`) since the original
query had
no `LIMIT` and therefore no way to iterate in batches.
Note: the `created_at` cursor used by `ListSources` is a pre-existing
design
in that SQL query and is left unchanged; timestamp collisions are
considered
acceptable at the batch sizes in use.
## Test plan
- [ ] New integration tests: `TestListSources`,
`TestListSources_ReturnsAllMatchingWithMixedClaims`,
`TestListSources_AnonymousReturnsAll`,
`TestListRegistries_ReturnsAllMatchingWithMixedClaims`,
`TestListRegistries_AnonymousReturnsAll`
- [ ] Smoke test "List completeness" scenarios: create 3 sources and 3
registries,
    verify all appear in the respective list endpoints
    
   Fixes #669 and #670

Signed-off-by: lujunsan <luisjuncaldev@gmail.com>

Author: lujunsan

.claude/skills/smoke-test/SKILL.md

@@ -170,6 +170,24 @@ Feature: Registry Server smoke tests
   Scenario: Publish with neither server nor skill in body returns 400
     When I POST /v1/entries with an empty payload
     Then the response status is 400
+
+  # ── List completeness ─────────────────────────────────────────────────────
+
+  Scenario: All created sources appear in list
+    Given three managed sources are created: smoke-src-a, smoke-src-b, smoke-src-c
+    When I GET /v1/sources
+    Then the response status is 200
+    And the body contains "smoke-src-a"
+    And the body contains "smoke-src-b"
+    And the body contains "smoke-src-c"
+
+  Scenario: All created registries appear in list
+    Given three registries are created referencing the managed sources
+    When I GET /v1/registries
+    Then the response status is 200
+    And the body contains "smoke-reg-a"
+    And the body contains "smoke-reg-b"
+    And the body contains "smoke-reg-c"
 ```
 
 ---
@@ -376,10 +394,39 @@ check "POST /v1/entries with both server+skill returns 400" 400 "$SC" "$(body)"
 SC=$(curl_post /v1/entries '{}')
 check "POST /v1/entries with neither server nor skill returns 400" 400 "$SC" "$(body)"
 
-# ─── Clean up managed source ──────────────────────────────────────────────────
+# ─── List completeness ───────────────────────────────────────────────────────
+echo ""
+echo "── List completeness ──"
+
+SC=$(curl_put /v1/sources/smoke-src-a '{"managed":{}}'); check "PUT /v1/sources/smoke-src-a returns 201" 201 "$SC" "$(body)" "smoke-src-a"
+SC=$(curl_put /v1/sources/smoke-src-b '{"managed":{}}'); check "PUT /v1/sources/smoke-src-b returns 201" 201 "$SC" "$(body)" "smoke-src-b"
+SC=$(curl_put /v1/sources/smoke-src-c '{"managed":{}}'); check "PUT /v1/sources/smoke-src-c returns 201" 201 "$SC" "$(body)" "smoke-src-c"
+
+SC=$(curl_get /v1/sources); BODY="$(body)"
+check "GET /v1/sources contains smoke-src-a" 200 "$SC" "$BODY" "smoke-src-a"
+check "GET /v1/sources contains smoke-src-b" 200 "$SC" "$BODY" "smoke-src-b"
+check "GET /v1/sources contains smoke-src-c" 200 "$SC" "$BODY" "smoke-src-c"
+
+SC=$(curl_put /v1/registries/smoke-reg-a '{"sources":["smoke-src-a"]}'); check "PUT /v1/registries/smoke-reg-a returns 201" 201 "$SC" "$(body)" "smoke-reg-a"
+SC=$(curl_put /v1/registries/smoke-reg-b '{"sources":["smoke-src-b"]}'); check "PUT /v1/registries/smoke-reg-b returns 201" 201 "$SC" "$(body)" "smoke-reg-b"
+SC=$(curl_put /v1/registries/smoke-reg-c '{"sources":["smoke-src-c"]}'); check "PUT /v1/registries/smoke-reg-c returns 201" 201 "$SC" "$(body)" "smoke-reg-c"
+
+SC=$(curl_get /v1/registries); BODY="$(body)"
+check "GET /v1/registries contains smoke-reg-a" 200 "$SC" "$BODY" "smoke-reg-a"
+check "GET /v1/registries contains smoke-reg-b" 200 "$SC" "$BODY" "smoke-reg-b"
+check "GET /v1/registries contains smoke-reg-c" 200 "$SC" "$BODY" "smoke-reg-c"
+
+# ─── Clean up ─────────────────────────────────────────────────────────────────
 SC=$(curl_del /v1/sources/managed-test)
 check "DELETE /v1/sources/managed-test returns 204" 204 "$SC" "$(body)"
 
+SC=$(curl_del /v1/registries/smoke-reg-a); check "DELETE /v1/registries/smoke-reg-a returns 204" 204 "$SC" "$(body)"
+SC=$(curl_del /v1/registries/smoke-reg-b); check "DELETE /v1/registries/smoke-reg-b returns 204" 204 "$SC" "$(body)"
+SC=$(curl_del /v1/registries/smoke-reg-c); check "DELETE /v1/registries/smoke-reg-c returns 204" 204 "$SC" "$(body)"
+SC=$(curl_del /v1/sources/smoke-src-a); check "DELETE /v1/sources/smoke-src-a returns 204" 204 "$SC" "$(body)"
+SC=$(curl_del /v1/sources/smoke-src-b); check "DELETE /v1/sources/smoke-src-b returns 204" 204 "$SC" "$(body)"
+SC=$(curl_del /v1/sources/smoke-src-c); check "DELETE /v1/sources/smoke-src-c returns 204" 204 "$SC" "$(body)"
+
 # ─── Summary ──────────────────────────────────────────────────────────────────
 echo ""
 echo "══════════════════════════════════════"

database/queries/registry.sql

@@ -2,7 +2,10 @@
 
 -- name: ListRegistries :many
 SELECT id, name, claims, creation_type, created_at, updated_at
-FROM registry ORDER BY name;
+FROM registry
+WHERE (sqlc.narg(cursor)::text IS NULL OR name > sqlc.narg(cursor))
+ORDER BY name
+LIMIT sqlc.arg(size)::bigint;
 
 -- name: GetRegistryByName :one
 SELECT id, name, claims, creation_type, created_at, updated_at

internal/db/sqlc/querier.go

@@ -484,6 +484,158 @@ func TestGetRegistryByName_HiddenWhenClaimsDontMatch(t *testing.T) {
 	assert.Nil(t, reg)
 }
 
+// ---------------------------------------------------------------------------
+// Mixed-claims listing tests — verify streaming batch loops return all
+// qualifying rows even when different claim values are interleaved.
+// ---------------------------------------------------------------------------
+
+func TestListSources_ReturnsAllMatchingWithMixedClaims(t *testing.T) {
+	t.Parallel()
+	svc, cleanup := setupTestServiceWithCodecs(t)
+	t.Cleanup(cleanup)
+
+	createCtx := t.Context()
+
+	// Create 4 sources: 2 with org=acme, 1 with org=contoso, 1 with no claims
+	_, err := svc.CreateSource(createCtx, "lsm-acme-1", managedSourceReq(map[string]any{"org": "acme"}))
+	require.NoError(t, err)
+
+	_, err = svc.CreateSource(createCtx, "lsm-acme-2", managedSourceReq(map[string]any{"org": "acme"}))
+	require.NoError(t, err)
+
+	_, err = svc.CreateSource(createCtx, "lsm-contoso", managedSourceReq(map[string]any{"org": "contoso"}))
+	require.NoError(t, err)
+
+	_, err = svc.CreateSource(createCtx, "lsm-open", managedSourceReq(nil))
+	require.NoError(t, err)
+
+	// List with acme JWT — should see both acme sources and the open (no-claims) source
+	listCtx := auth.ContextWithClaims(t.Context(), jwt.MapClaims{"org": "acme"})
+	sources, err := svc.ListSources(listCtx)
+	require.NoError(t, err)
+
+	names := sourceNames(sources)
+	assert.Contains(t, names, "lsm-acme-1")
+	assert.Contains(t, names, "lsm-acme-2")
+	assert.NotContains(t, names, "lsm-contoso")
+	assert.Contains(t, names, "lsm-open")
+}
+
+func TestListRegistries_ReturnsAllMatchingWithMixedClaims(t *testing.T) {
+	t.Parallel()
+	svc, cleanup := setupTestServiceWithCodecs(t)
+	t.Cleanup(cleanup)
+
+	createCtx := t.Context()
+
+	// Create 4 sources (one per registry) with different claims
+	createSourceForRegistry(t, svc, "lrm-acme-src-1", map[string]any{"org": "acme"})
+	createSourceForRegistry(t, svc, "lrm-acme-src-2", map[string]any{"org": "acme"})
+	createSourceForRegistry(t, svc, "lrm-contoso-src", map[string]any{"org": "contoso"})
+	createSourceForRegistry(t, svc, "lrm-open-src", nil)
+
+	// Create 4 registries with matching claims
+	_, err := svc.CreateRegistry(createCtx, "lrm-acme-reg-1", &service.RegistryCreateRequest{
+		Sources: []string{"lrm-acme-src-1"},
+		Claims:  map[string]any{"org": "acme"},
+	})
+	require.NoError(t, err)
+
+	_, err = svc.CreateRegistry(createCtx, "lrm-acme-reg-2", &service.RegistryCreateRequest{
+		Sources: []string{"lrm-acme-src-2"},
+		Claims:  map[string]any{"org": "acme"},
+	})
+	require.NoError(t, err)
+
+	_, err = svc.CreateRegistry(createCtx, "lrm-contoso-reg", &service.RegistryCreateRequest{
+		Sources: []string{"lrm-contoso-src"},
+		Claims:  map[string]any{"org": "contoso"},
+	})
+	require.NoError(t, err)
+
+	_, err = svc.CreateRegistry(createCtx, "lrm-open-reg", &service.RegistryCreateRequest{
+		Sources: []string{"lrm-open-src"},
+	})
+	require.NoError(t, err)
+
+	// List with acme JWT — should see both acme registries and the open one
+	listCtx := auth.ContextWithClaims(t.Context(), jwt.MapClaims{"org": "acme"})
+	registries, err := svc.ListRegistries(listCtx)
+	require.NoError(t, err)
+
+	names := registryNames(registries)
+	assert.Contains(t, names, "lrm-acme-reg-1")
+	assert.Contains(t, names, "lrm-acme-reg-2")
+	assert.NotContains(t, names, "lrm-contoso-reg")
+	assert.Contains(t, names, "lrm-open-reg")
+}
+
+func TestListSources_AnonymousReturnsAll(t *testing.T) {
+	t.Parallel()
+	svc, cleanup := setupTestServiceWithCodecs(t)
+	t.Cleanup(cleanup)
+
+	createCtx := t.Context()
+
+	// Create 3 sources with different claims
+	_, err := svc.CreateSource(createCtx, "lsa-acme", managedSourceReq(map[string]any{"org": "acme"}))
+	require.NoError(t, err)
+
+	_, err = svc.CreateSource(createCtx, "lsa-contoso", managedSourceReq(map[string]any{"org": "contoso"}))
+	require.NoError(t, err)
+
+	_, err = svc.CreateSource(createCtx, "lsa-open", managedSourceReq(nil))
+	require.NoError(t, err)
+
+	// List without JWT (anonymous) — should see all sources
+	sources, err := svc.ListSources(t.Context())
+	require.NoError(t, err)
+
+	names := sourceNames(sources)
+	assert.Contains(t, names, "lsa-acme")
+	assert.Contains(t, names, "lsa-contoso")
+	assert.Contains(t, names, "lsa-open")
+}
+
+func TestListRegistries_AnonymousReturnsAll(t *testing.T) {
+	t.Parallel()
+	svc, cleanup := setupTestServiceWithCodecs(t)
+	t.Cleanup(cleanup)
+
+	createCtx := t.Context()
+
+	// Create 3 sources + 3 registries with different claims
+	createSourceForRegistry(t, svc, "lra-acme-src", map[string]any{"org": "acme"})
+	createSourceForRegistry(t, svc, "lra-contoso-src", map[string]any{"org": "contoso"})
+	createSourceForRegistry(t, svc, "lra-open-src", nil)
+
+	_, err := svc.CreateRegistry(createCtx, "lra-acme-reg", &service.RegistryCreateRequest{
+		Sources: []string{"lra-acme-src"},
+		Claims:  map[string]any{"org": "acme"},
+	})
+	require.NoError(t, err)
+
+	_, err = svc.CreateRegistry(createCtx, "lra-contoso-reg", &service.RegistryCreateRequest{
+		Sources: []string{"lra-contoso-src"},
+		Claims:  map[string]any{"org": "contoso"},
+	})
+	require.NoError(t, err)
+
+	_, err = svc.CreateRegistry(createCtx, "lra-open-reg", &service.RegistryCreateRequest{
+		Sources: []string{"lra-open-src"},
+	})
+	require.NoError(t, err)
+
+	// List without JWT (anonymous) — should see all registries
+	registries, err := svc.ListRegistries(t.Context())
+	require.NoError(t, err)
+
+	names := registryNames(registries)
+	assert.Contains(t, names, "lra-acme-reg")
+	assert.Contains(t, names, "lra-contoso-reg")
+	assert.Contains(t, names, "lra-open-reg")
+}
+
 // ---------------------------------------------------------------------------
 // Helpers
 // ---------------------------------------------------------------------------

internal/db/sqlc/registry.sql.go

@@ -40,24 +40,16 @@ func (s *dbService) ListRegistries(ctx context.Context) ([]service.RegistryInfo,
 	}()
 
 	querier := sqlc.New(tx)
+	callerClaims := claimsFromCtx(ctx)
 
-	registries, err := querier.ListRegistries(ctx)
+	registries, err := streamRegistryRows(ctx, querier, callerClaims)
 	if err != nil {
 		otel.RecordError(span, err)
 		return nil, fmt.Errorf("failed to list registries: %w", err)
 	}
 
-	callerClaims := claimsFromCtx(ctx)
 	result := make([]service.RegistryInfo, 0, len(registries))
 	for _, reg := range registries {
-		// In authenticated mode, only return registries whose claims the caller covers
-		regClaims := db.DeserializeClaims(reg.Claims)
-		if callerClaims != nil {
-			if err := validateClaimsSubset(ctx, callerClaims, regClaims); err != nil {
-				continue
-			}
-		}
-
 		sources, err := querier.ListRegistrySources(ctx, reg.ID)
 		if err != nil {
 			otel.RecordError(span, err)
@@ -565,3 +557,40 @@ func unlinkAllSources(
 ) error {
 	return querier.UnlinkAllRegistrySources(ctx, registryID)
 }
+
+// streamRegistryRows fetches registry rows in batches, filtering by caller claims,
+// until the DB is exhausted. This avoids underfilled responses when post-filtering
+// drops rows that fail claims checks.
+func streamRegistryRows(
+	ctx context.Context,
+	querier *sqlc.Queries,
+	callerClaims map[string]any,
+) ([]sqlc.Registry, error) {
+	var accumulated []sqlc.Registry
+	params := sqlc.ListRegistriesParams{
+		Size: int64(service.MaxPageSize),
+	}
+
+	for {
+		batch, err := querier.ListRegistries(ctx, params)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, reg := range batch {
+			if err := validateClaimsSubsetBytes(ctx, callerClaims, reg.Claims); err != nil {
+				continue
+			}
+			accumulated = append(accumulated, reg)
+		}
+
+		if int64(len(batch)) < params.Size {
+			break
+		}
+
+		last := batch[len(batch)-1]
+		params.Cursor = &last.Name
+	}
+
+	return accumulated, nil
+}