bugfix indexer field

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)

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.
 //

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.