diff --git a/internal/api/server.go b/internal/api/server.go index 3ef5f8a..26e1fc5 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -351,6 +351,21 @@ func (s *Server) Start() error { // Republish events for all existing metadata go func() { + // Build any missing per-record field indexes first, in ONE + // keyspace pass. The republish below reads every record, and + // an un-indexed record read costs a full-keyspace SCAN — on a + // real store that walk warmed at ~1.3 records/s (a ~2.6h tail, + // paid again on every restart). Warming first turns the whole + // republish into index-served reads. Idempotent, so a + // fully-indexed store just pays the single scan. + // See storage/field_index.go. + started := time.Now() + if n, err := s.storage.WarmFieldIndexes(); err != nil { + log.Printf("[API] Warning: field index warm failed: %v (reads fall back to SCAN)", err) + } else if n > 0 { + log.Printf("[API] Warmed %d field indexes in %s", n, time.Since(started).Round(time.Millisecond)) + } + count, err := s.RepublishMetadata() if err != nil { log.Printf("[API] Warning: failed to republish metadata on startup: %v", err) diff --git a/internal/storage/field_index.go b/internal/storage/field_index.go index 98b995b..3892e5f 100644 --- a/internal/storage/field_index.go +++ b/internal/storage/field_index.go @@ -4,7 +4,9 @@ import ( "context" "fmt" "log" + "sort" "strings" + "time" "github.com/redis/go-redis/v9" ) @@ -265,6 +267,11 @@ func (c *Client) fieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix out = append(out, strings.TrimPrefix(f, fieldPrefix)) } } + // SMEMBERS order is arbitrary and SCAN order was too, so the CID array in + // GetMetadataDocument used to shuffle between calls. Sorting costs nothing + // at these sizes and makes the response byte-stable, which matters for + // anything diffing or caching it. + sort.Strings(out) return out, nil } @@ -285,9 +292,102 @@ func (c *Client) scanFieldsWithPrefixLocked(ctx context.Context, hashID, fieldPr break } } + sort.Strings(out) return out, nil } +// WarmFieldIndexes builds every missing field index in a SINGLE keyspace pass. +// +// Without it the indexes still appear — each record's first read falls back to +// a scan and backfills — but that transition is O(records x keyspace) and it +// runs again from scratch after any wipe. Measured on the real store (12,345 +// records / 445k keys) the lazy path warmed at ~1.3 records/s: a ~2.6 HOUR tail +// during which meta-core stays pinned exactly as it was before the fix. Worse, +// `RepublishMetadata` walks every record at startup, so that tail is paid on +// every restart. +// +// One bulk `file:*` scan grouped by record collapses it to a few seconds. This +// is the same shape `GetSearchIndexTuples` already uses for the bulk read, and +// for the same reason. +// +// Idempotent, and safe to run against a partially-indexed store: SADD of a +// member that is already there is a no-op. Returns the number of records it +// wrote an index for. +func (c *Client) WarmFieldIndexes() (int, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return 0, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + + filePrefix := c.buildKey("file:") // {filePrefix}{hashID}/{field} + indexKey := c.buildIndexKey() + + byRecord := make(map[string][]string) + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, filePrefix+"*", 1000).Result() + if err != nil { + return 0, fmt.Errorf("scan: %w", err) + } + for _, k := range keys { + if k == indexKey { + continue + } + // hashID has no '/'; field may contain them. + rest := strings.TrimPrefix(k, filePrefix) + slash := strings.IndexByte(rest, '/') + if slash < 0 { + continue + } + hashID, field := rest[:slash], rest[slash+1:] + if isReservedFieldName(field) { + // Already indexed — remember the record so we don't treat it + // as missing, but there is nothing to add for this key. + if _, seen := byRecord[hashID]; !seen { + byRecord[hashID] = nil + } + continue + } + byRecord[hashID] = append(byRecord[hashID], field) + } + cursor = next + if cursor == 0 { + break + } + } + + written := 0 + pipe := c.client.Pipeline() + pending := 0 + for hashID, fields := range byRecord { + if len(fields) == 0 { + continue + } + pipe.SAdd(ctx, c.buildFieldsKey(hashID), toAny(fields)...) + written++ + pending++ + // Bound the pipeline so a large store doesn't buffer every command. + if pending >= 500 { + if _, err := pipe.Exec(ctx); err != nil { + return written, fmt.Errorf("sadd pipeline: %w", err) + } + pipe = c.client.Pipeline() + pending = 0 + } + } + if pending > 0 { + if _, err := pipe.Exec(ctx); err != nil { + return written, fmt.Errorf("sadd pipeline: %w", err) + } + } + return written, nil +} + // recordHasFieldsLocked reports whether a record still has at least one // property key. Caller must hold c.mu. // diff --git a/internal/storage/field_index_test.go b/internal/storage/field_index_test.go index bb34075..1e85ba0 100644 --- a/internal/storage/field_index_test.go +++ b/internal/storage/field_index_test.go @@ -369,6 +369,68 @@ func TestFieldIndex_StaleEntryIsHarmless(t *testing.T) { }) } +// WarmFieldIndexes is what keeps an upgrade from spending hours reading +// un-indexed records one full-keyspace scan at a time. It must index every +// record in one pass, be idempotent, and leave already-indexed records alone. +func TestFieldIndex_WarmBuildsEveryMissingIndex(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + ctx := context.Background() + + // Two legacy records seeded straight into Redis (no index), plus one + // written through the normal path (already indexed). + legacy := map[string]map[string]string{ + "old1": {"filePath": "/a.mkv", "cids/bafkreiaaa": "true"}, + "old2": {"filePath": "/b.mkv", "titles/jpn": "ナルト", "deep/a/b": "x"}, + } + for root, fields := range legacy { + for f, v := range fields { + if err := c.client.Set(ctx, c.buildKeyPrefix(root)+f, v, 0).Err(); err != nil { + t.Fatalf("seed: %v", err) + } + } + if err := c.IndexAdd(root); err != nil { + t.Fatalf("IndexAdd: %v", err) + } + } + if err := c.SetMetadataFlat("fresh", map[string]string{"title": "x"}); err != nil { + t.Fatalf("SetMetadataFlat: %v", err) + } + + n, err := c.WarmFieldIndexes() + if err != nil { + t.Fatalf("WarmFieldIndexes: %v", err) + } + if n < 2 { + t.Errorf("warmed %d records, want at least the 2 legacy ones", n) + } + + if got := fieldNames(t, c, "old1"); !reflect.DeepEqual(got, []string{"cids/bafkreiaaa", "filePath"}) { + t.Errorf("old1 index = %v", got) + } + if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) { + t.Errorf("old2 index = %v", got) + } + if got := fieldNames(t, c, "fresh"); !reflect.DeepEqual(got, []string{"title"}) { + t.Errorf("fresh index disturbed = %v", got) + } + + // Reads must now agree with what the records actually hold. + for root, want := range legacy { + if got := mustGetMetadata(t, c, root); !reflect.DeepEqual(got, want) { + t.Errorf("%s read = %v, want %v", root, got, want) + } + } + + // Idempotent: a second pass changes nothing. + if _, err := c.WarmFieldIndexes(); err != nil { + t.Fatalf("second WarmFieldIndexes: %v", err) + } + if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) { + t.Errorf("old2 index after second warm = %v", got) + } + }) +} + // The bulk search read enumerates the whole file: keyspace and MGETs what it // finds. The index keys are SETs living in that same namespace, so it has to // skip them or the MGET fails with WRONGTYPE and search dies wholesale.