From a61057cd815795e5f3bf512bf08951a56cc1c2b1 Mon Sep 17 00:00:00 2001 From: Worph Date: Mon, 3 Aug 2026 18:42:05 +0200 Subject: [PATCH] optimisation on assets retreive --- internal/events/meta_publisher.go | 10 + internal/storage/cid_resolution.go | 34 +-- internal/storage/client.go | 168 ++++++----- internal/storage/field_index.go | 386 +++++++++++++++++++++++++ internal/storage/field_index_test.go | 402 +++++++++++++++++++++++++++ 5 files changed, 909 insertions(+), 91 deletions(-) create mode 100644 internal/storage/field_index.go create mode 100644 internal/storage/field_index_test.go diff --git a/internal/events/meta_publisher.go b/internal/events/meta_publisher.go index 4967bad..69d7d5f 100644 --- a/internal/events/meta_publisher.go +++ b/internal/events/meta_publisher.go @@ -9,6 +9,8 @@ import ( "time" "github.com/redis/go-redis/v9" + + "github.com/metazla/meta-core/internal/storage" ) const ( @@ -126,6 +128,14 @@ func (p *MetaPublisher) publishEvent(channel, operation string) { if strings.Contains(key, "__index__") || !strings.HasPrefix(key, "file:") { return } + // Skip the per-record field index. It is bookkeeping (see + // storage.FieldsField), not a metadata field — publishing it would make + // every downstream consumer that parses `file:/` off this + // stream (meta-search, meta-dup, meta-fuse, meta-sort) see a phantom + // field change on a key they can never read. + if strings.HasSuffix(key, "/"+storage.FieldsField) { + return + } // Build event fields fields := map[string]interface{}{ diff --git a/internal/storage/cid_resolution.go b/internal/storage/cid_resolution.go index 73314c6..bc7c034 100644 --- a/internal/storage/cid_resolution.go +++ b/internal/storage/cid_resolution.go @@ -83,6 +83,7 @@ func (c *Client) Mint(filePath string, size, mtimeNano int64) (string, error) { pipe.Set(ctx, prefix+"sizeByte", strconv.FormatInt(size, 10), 0) pipe.Set(ctx, prefix+"mtimeNano", strconv.FormatInt(mtimeNano, 10), 0) pipe.SAdd(ctx, c.buildIndexKey(), uuid) + c.addFieldsToPipe(ctx, pipe, uuid, []string{"filePath", "sizeByte", "mtimeNano"}) if _, err := pipe.Exec(ctx); err != nil { return "", fmt.Errorf("mint pipeline: %w", err) } @@ -116,7 +117,13 @@ func (c *Client) addAliasLocked(ctx context.Context, uuid, cidStr string) error return fmt.Errorf("uuid and cid are required") } - keysetKey := c.buildKeyPrefix(uuid) + CIDsKeyPrefix + cidStr + keysetField := CIDsKeyPrefix + cidStr + keysetKey := c.buildKeyPrefix(uuid) + keysetField + + // Every branch below writes the key-set member, so index the field name + // once here — including the two guard branches that skip the alias itself. + // Missing it would leave `cids/` invisible to a record read. + defer c.addFieldLocked(ctx, uuid, keysetField) // Self-pointing alias guard. If the caller is trying to register // cid: → (i.e. the root IS the cid value), that's the @@ -233,26 +240,13 @@ func (c *Client) AddDuplicatePath(uuid, path string) (bool, error) { return added > 0, nil } -// cidsForRootLocked scans the bare-CID key-set for a root and returns its -// members (bare CIDs, with the cids/ prefix stripped). Caller must hold c.mu. +// cidsForRootLocked returns a root's bare-CID key-set members (with the cids/ +// prefix stripped). Caller must hold c.mu. +// +// Reads the field index and filters by prefix; only an un-indexed root falls +// back to a SCAN, and then over the narrower cids/ glob. See field_index.go. func (c *Client) cidsForRootLocked(ctx context.Context, uuid string) ([]string, error) { - prefix := c.buildKeyPrefix(uuid) + CIDsKeyPrefix - var out []string - var cursor uint64 - for { - keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() - if err != nil { - return nil, fmt.Errorf("scan cids: %w", err) - } - for _, k := range keys { - out = append(out, strings.TrimPrefix(k, prefix)) - } - cursor = next - if cursor == 0 { - break - } - } - return out, nil + return c.fieldsWithPrefixLocked(ctx, uuid, CIDsKeyPrefix) } // DeleteRoot removes everything keyed off uuid: every file:/* property diff --git a/internal/storage/client.go b/internal/storage/client.go index e908363..d1799cc 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -109,8 +109,12 @@ func (c *Client) buildIndexKey() string { return c.buildKey("file:__index__") } -// GetMetadataFlat retrieves all metadata for a file as a flat map -// Uses flat keys: SCAN file:{hashId}/* then MGET +// GetMetadataFlat retrieves all metadata for a file as a flat map. +// +// Reads the record's field-name index and MGETs those keys — two round-trips, +// independent of keyspace size. Records written before the index existed fall +// back to the old SCAN and are indexed on the way out. See field_index.go for +// why the index exists and why it is safe to derive. func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -122,41 +126,7 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - prefix := c.buildKeyPrefix(hashID) - result := make(map[string]string) - - // Scan for all keys with this prefix - var cursor uint64 - for { - keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() - if err != nil { - return nil, fmt.Errorf("scan failed: %w", err) - } - - if len(keys) > 0 { - values, err := c.client.MGet(ctx, keys...).Result() - if err != nil { - return nil, fmt.Errorf("mget failed: %w", err) - } - for i, key := range keys { - field := strings.TrimPrefix(key, prefix) - if values[i] != nil { - result[field] = values[i].(string) - } - } - } - - cursor = nextCursor - if cursor == 0 { - break - } - } - - if len(result) == 0 { - return nil, nil - } - - return result, nil + return c.readMetadataLocked(ctx, hashID) } // SetMetadataFlat stores metadata for a file using flat keys @@ -179,9 +149,21 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro prefix := c.buildKeyPrefix(hashID) pipe := c.client.Pipeline() + fields := make([]string, 0, len(metadata)) for field, value := range metadata { + // A payload carrying the reserved name (a hand-made snapshot, a buggy + // client) would SET a string over the index SET and break every + // subsequent read of this record. Drop it — it is derived state, so + // there is never a reason to accept it from outside. + if isReservedFieldName(field) { + continue + } pipe.Set(ctx, prefix+field, value, 0) + fields = append(fields, field) } + // Same pipeline as the SETs, so indexing the field names costs no extra + // round-trip. See field_index.go. + c.addFieldsToPipe(ctx, pipe, hashID, fields) if _, err := pipe.Exec(ctx); err != nil { return fmt.Errorf("pipeline exec failed: %w", err) @@ -256,6 +238,12 @@ func (c *Client) SetProperty(hashID, property, value string) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() + // See SetMetadataFlat: the reserved name addresses the derived index SET, + // not a metadata field. + if isReservedFieldName(property) { + return fmt.Errorf("field name %q is reserved", property) + } + key := c.buildKeyPrefix(hashID) + property // Use SET to set the key @@ -263,6 +251,8 @@ func (c *Client) SetProperty(hashID, property, value string) error { return fmt.Errorf("set failed: %w", err) } + c.addFieldLocked(ctx, hashID, property) + // Add to index set if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil { return fmt.Errorf("sadd index failed: %w", err) @@ -387,32 +377,14 @@ func normalizeFilesRelativePath(p string) string { // getMetadataFlatInternal mirrors GetMetadataFlat without re-acquiring the lock. // Callers must already hold c.mu. func (c *Client) getMetadataFlatInternal(ctx context.Context, hashID string) (map[string]string, error) { - prefix := c.buildKeyPrefix(hashID) - result := make(map[string]string) - - var cursor uint64 - for { - keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() - if err != nil { - return nil, fmt.Errorf("scan failed: %w", err) - } - if len(keys) > 0 { - values, err := c.client.MGet(ctx, keys...).Result() - if err != nil { - return nil, fmt.Errorf("mget failed: %w", err) - } - for i, key := range keys { - if values[i] == nil { - continue - } - field := strings.TrimPrefix(key, prefix) - result[field] = values[i].(string) - } - } - cursor = nextCursor - if cursor == 0 { - break - } + result, err := c.readMetadataLocked(ctx, hashID) + if err != nil { + return nil, err + } + // Unlike GetMetadataFlat, this variant hands back an empty map rather than + // nil for a record with no fields — GetMetadataDocument writes into it. + if result == nil { + result = make(map[string]string) } return result, nil } @@ -449,9 +421,17 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i prefix := c.buildKeyPrefix(hashID) pipe := c.client.Pipeline() + fields := make([]string, 0, len(metadata)) for field, value := range metadata { + // See SetMetadataFlat: never let an inbound payload overwrite the + // derived index SET with a string. + if isReservedFieldName(field) { + continue + } pipe.Set(ctx, prefix+field, value, 0) + fields = append(fields, field) } + c.addFieldsToPipe(ctx, pipe, hashID, fields) if _, err := pipe.Exec(ctx); err != nil { return 0, fmt.Errorf("pipeline exec failed: %w", err) @@ -484,7 +464,11 @@ func (c *Client) DeleteProperty(hashID, property string) error { defer cancel() key := c.buildKeyPrefix(hashID) + property - return c.client.Del(ctx, key).Err() + if err := c.client.Del(ctx, key).Err(); err != nil { + return err + } + c.removeFieldLocked(ctx, hashID, property) + return nil } // AddToSet adds a value to a set-type field (stored as pipe-delimited string in flat key) @@ -530,6 +514,9 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) { return false, err } + // The SET above may have created the key, so the field name is new. + c.addFieldLocked(ctx, hashID, property) + // Add to index set if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil { return true, fmt.Errorf("sadd index failed: %w", err) @@ -582,7 +569,12 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { // Save back (or delete key if empty) if len(newValues) == 0 { - return true, c.client.Del(ctx, key).Err() + if err := c.client.Del(ctx, key).Err(); err != nil { + return true, err + } + // The key is gone, so the field name must leave the index too. + c.removeFieldLocked(ctx, hashID, property) + return true, nil } newValue := strings.Join(newValues, "|") @@ -764,6 +756,12 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) { if k == indexKey { continue } + // Per-record field indexes live under the same prefix and are SETs + // — MGET would error on them, and they are bookkeeping, not + // searchable metadata. + if strings.HasSuffix(k, "/"+FieldsField) { + continue + } keyBatch = append(keyBatch, k) if len(keyBatch) >= 1000 { if err := flush(); err != nil { @@ -1039,6 +1037,12 @@ func (c *Client) FindByValue(contains string, fields []string, limit int) ([]Fin if field == "" { continue } + // The per-record field index is a SET, so an MGET over those keys + // would fail the whole query with WRONGTYPE. It holds field names, + // not values, so there is nothing here to search anyway. + if isReservedFieldName(field) { + continue + } if len(matches) >= limit { truncated = true break @@ -1207,7 +1211,17 @@ func (c *Client) SetRawValue(key, value string) error { if t != "none" && t != "string" { return fmt.Errorf("key has incompatible type %q (only string is editable)", t) } - return c.client.Set(ctx, full, value, 0).Err() + if err := c.client.Set(ctx, full, value, 0).Err(); err != nil { + return err + } + // The raw editor can mint a record property out of band, so it has to keep + // the field index honest like every other write path. Non-record keys + // (`cid:*`, `udl:*`, `file:__index__`, the index set itself) split false + // and are left alone. + if hashID, field, ok := c.splitRecordKey(full); ok { + c.addFieldLocked(ctx, hashID, field) + } + return nil } // DeleteRawKey deletes the exact key. Returns true if the key existed. @@ -1222,10 +1236,16 @@ func (c *Client) DeleteRawKey(key string) (bool, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - n, err := c.client.Del(ctx, c.buildKey(key)).Result() + full := c.buildKey(key) + n, err := c.client.Del(ctx, full).Result() if err != nil { return false, err } + if n > 0 { + if hashID, field, ok := c.splitRecordKey(full); ok { + c.removeFieldLocked(ctx, hashID, field) + } + } return n > 0, nil } @@ -1257,14 +1277,20 @@ func (c *Client) IndexRemoveIfEmpty(hashID string) (bool, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - prefix := c.buildKeyPrefix(hashID) - keys, _, err := c.client.Scan(ctx, 0, prefix+"*", 1).Result() + nonEmpty, err := c.recordHasFieldsLocked(ctx, hashID) if err != nil { - return false, fmt.Errorf("scan failed: %w", err) + return false, err } - if len(keys) > 0 { + if nonEmpty { return false, nil } + + // Nothing left but (possibly) an empty field index — drop it too, so a + // de-indexed root leaves no orphan key behind. + if err := c.client.Del(ctx, c.buildFieldsKey(hashID)).Err(); err != nil { + return false, fmt.Errorf("del field index: %w", err) + } + removed, err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Result() if err != nil { return false, fmt.Errorf("srem failed: %w", err) diff --git a/internal/storage/field_index.go b/internal/storage/field_index.go new file mode 100644 index 0000000..98b995b --- /dev/null +++ b/internal/storage/field_index.go @@ -0,0 +1,386 @@ +package storage + +import ( + "context" + "fmt" + "log" + "strings" + + "github.com/redis/go-redis/v9" +) + +// The per-record field-name index. +// +// A record is stored as one flat Redis STRING per leaf +// (`file:/`), and that shape is deliberate — it is what lets +// keyspace notifications fire per *field*, so meta-fuse / meta-editor / +// meta-search can react to one field changing instead of diffing a whole hash +// (see docs/project-architecture/persistent-storage.md, and +// dev/test/suites/flat-keys.bats which asserts records are never hashes). +// +// The cost of that shape was that reading one record meant asking Redis "which +// keys start with `file:/`?", and Redis has no prefix index — `SCAN +// MATCH` walks the ENTIRE keyspace. On a real store (445k keys, ~36 fields per +// record) a single record read cost ~670 SCAN round-trips and ~0.5s of Redis +// CPU; SCAN accounted for ~97.5% of all Redis CPU time on the box. That sat on +// the byte path for every poster meta-share served, which is how it surfaced. +// +// So we keep the flat keys and index their *names*: +// +// file:/__fields__ SET{ "filePath", "sizeByte", "cids/bafk…", … } +// +// A read becomes SMEMBERS + a chunked MGET — two round-trips instead of ~670. +// This mirrors `file:__index__` (the set of every root) one level down, and the +// bulk reads already worked this way: `GetTuplesForAllFiles` composes its keys +// from the index rather than scanning per root. +// +// **The set is derived, never authoritative.** The flat keys remain the source +// of truth. A record whose set is missing (anything written before this landed) +// falls back to the old SCAN and populates the set on the way out, so existing +// deployments heal themselves on first touch with no migration and no +// SchemaVersion bump — which matters, because a bump is a flag day that makes +// operators wipe Redis (see schema_version.go). +// +// The consistency argument that makes a derived index safe here is meta-core's +// single-writer rule: leader election publishes an empty `redisUrl` on purpose +// (internal/leader/election.go), so nothing outside this process can write a +// `file:` key behind the index's back. + +// FieldsField is the reserved field name holding a record's field-name index. +// +// It is a Redis SET living at `file:/__fields__`, i.e. inside the same +// prefix as the record's own fields. That keeps every existing sweep correct +// for free — `DeleteMetadata`, `DeleteRoot` and `ClearAllMetadata` all +// SCAN+DEL `file:/*` and therefore drop the index along with the record. +// The flip side is that every reader which enumerates that prefix has to skip +// it, exactly as they already skip `file:__index__`. +const FieldsField = "__fields__" + +// mgetBatch caps how many keys go into one MGET. Redis limits arguments per +// command; 1000 is the batch size `GetTuplesForAllFiles` already uses. +const mgetBatch = 1000 + +// buildFieldsKey constructs the key for a record's field-name index set. +func (c *Client) buildFieldsKey(hashID string) string { + return c.buildKeyPrefix(hashID) + FieldsField +} + +// isReservedFieldName reports whether a field name is bookkeeping rather than +// metadata, and so must never be returned to an API caller, exported into a +// snapshot, or folded into the search haystack. +func isReservedFieldName(field string) bool { + return field == FieldsField +} + +// readMetadataLocked returns a record's full flat metadata. Caller must hold +// c.mu (either mode). +// +// Fast path: SMEMBERS the field index, then MGET those keys. Fallback: the +// legacy full-keyspace SCAN, whose result is used to populate the index so the +// next read takes the fast path. +// +// Returns (nil, nil) for a record with no fields — callers treat that as 404. +func (c *Client) readMetadataLocked(ctx context.Context, hashID string) (map[string]string, error) { + fields, err := c.client.SMembers(ctx, c.buildFieldsKey(hashID)).Result() + if err != nil && err != redis.Nil { + return nil, fmt.Errorf("smembers fields: %w", err) + } + + if len(fields) > 0 { + return c.mgetFieldsLocked(ctx, hashID, fields) + } + + // No index yet: a record written before the index existed, or one that + // genuinely has no fields. Fall back to the scan, then backfill. + result, scanned, err := c.scanMetadataLocked(ctx, hashID) + if err != nil { + return nil, err + } + if len(scanned) > 0 { + // Best-effort: a failed backfill costs the next read a scan, it does + // not make the read wrong. Never fail the caller's read for it. + if err := c.client.SAdd(ctx, c.buildFieldsKey(hashID), toAny(scanned)...).Err(); err != nil { + logFieldIndexBackfillFailure(hashID, err) + } + } + if len(result) == 0 { + return nil, nil + } + return result, nil +} + +// mgetFieldsLocked reads the named fields of a record in chunked MGETs. +// +// A field named in the index whose key is gone reads back as nil and is simply +// skipped: a *stale* index entry is harmless. (A *missing* one would silently +// drop data, which is why every write path maintains the set.) +func (c *Client) mgetFieldsLocked(ctx context.Context, hashID string, fields []string) (map[string]string, error) { + prefix := c.buildKeyPrefix(hashID) + + // Keep the reserved name out of the key list AND out of the result — it is + // a SET, so MGET would error on it in real Redis rather than return nil. + wanted := make([]string, 0, len(fields)) + for _, f := range fields { + if !isReservedFieldName(f) { + wanted = append(wanted, f) + } + } + if len(wanted) == 0 { + return nil, nil + } + + result := make(map[string]string, len(wanted)) + for start := 0; start < len(wanted); start += mgetBatch { + end := start + mgetBatch + if end > len(wanted) { + end = len(wanted) + } + chunk := wanted[start:end] + keys := make([]string, len(chunk)) + for i, f := range chunk { + keys[i] = prefix + f + } + values, err := c.client.MGet(ctx, keys...).Result() + if err != nil { + return nil, fmt.Errorf("mget failed: %w", err) + } + for i, v := range values { + if s, ok := v.(string); ok { + result[chunk[i]] = s + } + } + } + + if len(result) == 0 { + return nil, nil + } + return result, nil +} + +// scanMetadataLocked is the pre-index read: SCAN the record's key prefix, then +// MGET. Retained as the fallback for un-indexed records and as the source for +// the backfill. Returns the metadata plus the field names it saw (which +// includes fields whose value came back nil, so the index stays complete). +func (c *Client) scanMetadataLocked(ctx context.Context, hashID string) (map[string]string, []string, error) { + prefix := c.buildKeyPrefix(hashID) + result := make(map[string]string) + var fields []string + + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return nil, nil, fmt.Errorf("scan failed: %w", err) + } + + if len(keys) > 0 { + wanted := make([]string, 0, len(keys)) + for _, key := range keys { + field := strings.TrimPrefix(key, prefix) + if isReservedFieldName(field) { + continue + } + fields = append(fields, field) + wanted = append(wanted, key) + } + if len(wanted) > 0 { + values, err := c.client.MGet(ctx, wanted...).Result() + if err != nil { + return nil, nil, fmt.Errorf("mget failed: %w", err) + } + for i, key := range wanted { + if s, ok := values[i].(string); ok { + result[strings.TrimPrefix(key, prefix)] = s + } + } + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + return result, fields, nil +} + +// addFieldsToPipe records field names in the index as part of an existing +// pipeline, so a multi-field write stays one round-trip. +func (c *Client) addFieldsToPipe(ctx context.Context, pipe redis.Pipeliner, hashID string, fields []string) { + names := filterReserved(fields) + if len(names) == 0 { + return + } + pipe.SAdd(ctx, c.buildFieldsKey(hashID), toAny(names)...) +} + +// addFieldLocked records a single field name in the index. Caller must hold +// c.mu. Best-effort by the same argument as the backfill: the flat key is the +// source of truth, and a lost index entry is repaired by the next scan +// fallback — so it must not fail the write it accompanies. +func (c *Client) addFieldLocked(ctx context.Context, hashID, field string) { + if isReservedFieldName(field) { + return + } + if err := c.client.SAdd(ctx, c.buildFieldsKey(hashID), field).Err(); err != nil { + logFieldIndexWriteFailure(hashID, field, err) + } +} + +// removeFieldLocked drops a field name from the index. Caller must hold c.mu. +// +// A stale entry left behind by a failure is harmless (the MGET reads nil and +// skips it), so this is best-effort too. +func (c *Client) removeFieldLocked(ctx context.Context, hashID, field string) { + if isReservedFieldName(field) { + return + } + if err := c.client.SRem(ctx, c.buildFieldsKey(hashID), field).Err(); err != nil { + logFieldIndexWriteFailure(hashID, field, err) + } +} + +// fieldsWithPrefixLocked returns the record's indexed field names under a +// field-name prefix (e.g. "cids/"), falling back to a SCAN when the record has +// no index yet. Caller must hold c.mu. +func (c *Client) fieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix string) ([]string, error) { + fields, err := c.client.SMembers(ctx, c.buildFieldsKey(hashID)).Result() + if err != nil && err != redis.Nil { + return nil, fmt.Errorf("smembers fields: %w", err) + } + if len(fields) == 0 { + // Un-indexed record: scan the narrower `*` glob + // rather than the whole record, and do not backfill from it — a + // partial view must never be mistaken for a complete index. + return c.scanFieldsWithPrefixLocked(ctx, hashID, fieldPrefix) + } + + var out []string + for _, f := range fields { + if isReservedFieldName(f) { + continue + } + if strings.HasPrefix(f, fieldPrefix) { + out = append(out, strings.TrimPrefix(f, fieldPrefix)) + } + } + return out, nil +} + +func (c *Client) scanFieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix string) ([]string, error) { + prefix := c.buildKeyPrefix(hashID) + fieldPrefix + var out []string + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan fields: %w", err) + } + for _, k := range keys { + out = append(out, strings.TrimPrefix(k, prefix)) + } + cursor = next + if cursor == 0 { + break + } + } + return out, nil +} + +// recordHasFieldsLocked reports whether a record still has at least one +// property key. Caller must hold c.mu. +// +// `__fields__` must not count — it lives under the record's own prefix, so a +// naive "does any `file:/*` key exist?" probe would answer yes for a record +// whose every property has been deleted, and the root could never be +// de-indexed. +func (c *Client) recordHasFieldsLocked(ctx context.Context, hashID string) (bool, error) { + n, err := c.client.SCard(ctx, c.buildFieldsKey(hashID)).Result() + if err != nil && err != redis.Nil { + return false, fmt.Errorf("scard fields: %w", err) + } + if n > 0 { + return true, nil + } + + // Either genuinely empty, or an un-indexed record from before the index + // existed — the scan is the only way to tell those apart. + prefix := c.buildKeyPrefix(hashID) + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return false, fmt.Errorf("scan failed: %w", err) + } + for _, k := range keys { + if !isReservedFieldName(strings.TrimPrefix(k, prefix)) { + return true, nil + } + } + cursor = next + if cursor == 0 { + return false, nil + } + } +} + +// splitRecordKey decomposes a full Redis key into (hashID, field) when it is a +// record property key. Reports ok=false for anything else — the index set +// itself, `file:__index__`, and any key outside the `file:` namespace. +// +// This is what lets the raw KV escape hatch (SetRawValue / DeleteRawKey) keep +// the index honest without its callers knowing the layout. +func (c *Client) splitRecordKey(key string) (hashID, field string, ok bool) { + rest := key + if c.prefix != "" { + var found bool + rest, found = strings.CutPrefix(rest, c.prefix) + if !found { + return "", "", false + } + } + rest, found := strings.CutPrefix(rest, "file:") + if !found { + return "", "", false + } + hashID, field, found = strings.Cut(rest, "/") + if !found || hashID == "" || field == "" { + // `file:__index__` has no slash and lands here. + return "", "", false + } + if isReservedFieldName(field) { + return "", "", false + } + return hashID, field, true +} + +// The index is derived and self-healing, so a write failure degrades +// performance (the next read falls back to a scan and re-populates) rather +// than correctness. Log it rather than failing the caller's write — but log it +// loudly, because a *persistent* failure means every read is scanning again. +func logFieldIndexWriteFailure(hashID, field string, err error) { + log.Printf("[Storage] field index update failed for %s/%s: %v (reads fall back to SCAN)", hashID, field, err) +} + +func logFieldIndexBackfillFailure(hashID string, err error) { + log.Printf("[Storage] field index backfill failed for %s: %v (reads keep falling back to SCAN)", hashID, err) +} + +func filterReserved(fields []string) []string { + out := make([]string, 0, len(fields)) + for _, f := range fields { + if !isReservedFieldName(f) { + out = append(out, f) + } + } + return out +} + +func toAny(ss []string) []interface{} { + out := make([]interface{}, len(ss)) + for i, s := range ss { + out[i] = s + } + return out +} diff --git a/internal/storage/field_index_test.go b/internal/storage/field_index_test.go new file mode 100644 index 0000000..bb34075 --- /dev/null +++ b/internal/storage/field_index_test.go @@ -0,0 +1,402 @@ +package storage + +import ( + "context" + "reflect" + "sort" + "testing" +) + +// Tests for the per-record field-name index (field_index.go). +// +// The index is what makes a record read O(fields) instead of O(keyspace). It is +// derived state, so the properties worth pinning are: (1) a read served from +// the index returns exactly what a scan would have, (2) a record with no index +// still reads correctly and gets one, and (3) every write path keeps it in +// sync — because a *missing* entry silently drops a field from every +// subsequent read, which is the one failure mode that would be invisible. +// +// Every case runs with and without a namespace prefix, matching clear_test.go. + +func forEachPrefix(t *testing.T, fn func(t *testing.T, c *Client)) { + t.Helper() + for _, prefix := range []string{"", "mm:"} { + name := "no-prefix" + if prefix != "" { + name = "prefix-" + prefix + } + t.Run(name, func(t *testing.T) { + c, _ := newTestClient(t, prefix) + fn(t, c) + }) + } +} + +func fieldNames(t *testing.T, c *Client, hashID string) []string { + t.Helper() + got, err := c.client.SMembers(context.Background(), c.buildFieldsKey(hashID)).Result() + if err != nil { + t.Fatalf("smembers fields: %v", err) + } + sort.Strings(got) + return got +} + +func mustGetMetadata(t *testing.T, c *Client, hashID string) map[string]string { + t.Helper() + got, err := c.GetMetadataFlat(hashID) + if err != nil { + t.Fatalf("GetMetadataFlat: %v", err) + } + return got +} + +// SetMetadataFlat must index every field it writes, and the read must come back +// through the index rather than a scan. +func TestFieldIndex_SetMetadataFlatIndexesFields(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + want := map[string]string{ + "filePath": "/files/watch/a.mkv", + "sizeByte": "123", + "cids/bafkreiaaa": "true", + "titles/jpn": "ナルト", + "categories/TV/UHD": "true", + "info/files/0/path": "a.mkv", + "provenance": `{"source":"gateway"}`, + "plot/eng": "a plot with / slashes", + "stream/3": "eng", + "genres/action": "true", + "originaltitle": "Naruto", + "deeply/nested/field/name/here": "x", + } + if err := c.SetMetadataFlat("root1", want); err != nil { + t.Fatalf("SetMetadataFlat: %v", err) + } + + if got := mustGetMetadata(t, c, "root1"); !reflect.DeepEqual(got, want) { + t.Errorf("GetMetadataFlat mismatch\n got %v\nwant %v", got, want) + } + + // The index must name every field — including the ones containing '/', + // which is where a naive "split on the first slash" would go wrong. + wantNames := make([]string, 0, len(want)) + for k := range want { + wantNames = append(wantNames, k) + } + sort.Strings(wantNames) + if got := fieldNames(t, c, "root1"); !reflect.DeepEqual(got, wantNames) { + t.Errorf("field index mismatch\n got %v\nwant %v", got, wantNames) + } + }) +} + +// A record written before the index existed has flat keys but no set. The read +// must still be correct, and must leave an index behind so the next read takes +// the fast path. This is the migration path for the ~12k records already on +// disk: no flag day, no SchemaVersion bump. +func TestFieldIndex_LazyBackfillForUnindexedRecord(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + ctx := context.Background() + prefix := c.buildKeyPrefix("legacy") + want := map[string]string{ + "filePath": "/files/watch/old.mkv", + "sizeByte": "42", + "cids/bafkreiold": "true", + } + // Write the flat keys directly, bypassing every index-maintaining path + // — exactly the shape a pre-upgrade store is in. + for f, v := range want { + if err := c.client.Set(ctx, prefix+f, v, 0).Err(); err != nil { + t.Fatalf("seed %s: %v", f, err) + } + } + if names := fieldNames(t, c, "legacy"); len(names) != 0 { + t.Fatalf("precondition: expected no index, got %v", names) + } + + if got := mustGetMetadata(t, c, "legacy"); !reflect.DeepEqual(got, want) { + t.Errorf("scan-fallback read mismatch\n got %v\nwant %v", got, want) + } + + wantNames := []string{"cids/bafkreiold", "filePath", "sizeByte"} + if got := fieldNames(t, c, "legacy"); !reflect.DeepEqual(got, wantNames) { + t.Errorf("backfill mismatch\n got %v\nwant %v", got, wantNames) + } + + // Second read, now served from the index, must agree with the first. + if got := mustGetMetadata(t, c, "legacy"); !reflect.DeepEqual(got, want) { + t.Errorf("indexed read disagrees with scan read\n got %v\nwant %v", got, want) + } + }) +} + +// Each single-field write path must add to the index, and each delete path must +// remove from it. A field written but not indexed vanishes from every later +// read — silently — so this is the property that matters most. +func TestFieldIndex_SingleFieldWritePathsStayInSync(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + if err := c.SetProperty("root2", "title", "Naruto"); err != nil { + t.Fatalf("SetProperty: %v", err) + } + if _, err := c.AddToSet("root2", "tags", "anime"); err != nil { + t.Fatalf("AddToSet: %v", err) + } + if _, err := c.MergeMetadataFlat("root2", map[string]string{"year": "2002"}); err != nil { + t.Fatalf("MergeMetadataFlat: %v", err) + } + + want := map[string]string{"title": "Naruto", "tags": "anime", "year": "2002"} + if got := mustGetMetadata(t, c, "root2"); !reflect.DeepEqual(got, want) { + t.Errorf("after writes\n got %v\nwant %v", got, want) + } + + // DeleteProperty drops the key and the index entry. + if err := c.DeleteProperty("root2", "title"); err != nil { + t.Fatalf("DeleteProperty: %v", err) + } + // RemoveFromSet deletes the key once its last member is gone. + if _, err := c.RemoveFromSet("root2", "tags", "anime"); err != nil { + t.Fatalf("RemoveFromSet: %v", err) + } + + want = map[string]string{"year": "2002"} + if got := mustGetMetadata(t, c, "root2"); !reflect.DeepEqual(got, want) { + t.Errorf("after deletes\n got %v\nwant %v", got, want) + } + if got := fieldNames(t, c, "root2"); !reflect.DeepEqual(got, []string{"year"}) { + t.Errorf("index after deletes = %v, want [year]", got) + } + }) +} + +// The raw KV editor can mint or drop a record property out of band. It is the +// one write path whose caller does not know the record layout, so the storage +// layer has to keep the index honest for it. +func TestFieldIndex_RawKeyEditorStaysInSync(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + if err := c.SetRawValue("file:root3/hand-written", "v"); err != nil { + t.Fatalf("SetRawValue: %v", err) + } + if got := fieldNames(t, c, "root3"); !reflect.DeepEqual(got, []string{"hand-written"}) { + t.Errorf("index after SetRawValue = %v", got) + } + if got := mustGetMetadata(t, c, "root3"); !reflect.DeepEqual(got, map[string]string{"hand-written": "v"}) { + t.Errorf("read after SetRawValue = %v", got) + } + + if _, err := c.DeleteRawKey("file:root3/hand-written"); err != nil { + t.Fatalf("DeleteRawKey: %v", err) + } + if got := fieldNames(t, c, "root3"); len(got) != 0 { + t.Errorf("index after DeleteRawKey = %v, want empty", got) + } + + // Keys outside the record namespace must be left alone entirely. + if err := c.SetRawValue("cid:bafkreiaaa", "root3"); err != nil { + t.Fatalf("SetRawValue non-record: %v", err) + } + if _, _, ok := c.splitRecordKey(c.buildKey("cid:bafkreiaaa")); ok { + t.Error("cid: key was treated as a record property") + } + if _, _, ok := c.splitRecordKey(c.buildIndexKey()); ok { + t.Error("file:__index__ was treated as a record property") + } + }) +} + +// Mint and AddAlias write fields through their own pipelines, so they need the +// index too. AddAlias in particular is how `cids/` members appear — miss +// it and a record's CID key-set becomes invisible to every read. +func TestFieldIndex_MintAndAliasIndexTheirFields(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + uuid, err := c.Mint("/files/watch/x.mkv", 10, 20) + if err != nil { + t.Fatalf("Mint: %v", err) + } + if got := fieldNames(t, c, uuid); !reflect.DeepEqual(got, []string{"filePath", "mtimeNano", "sizeByte"}) { + t.Errorf("index after Mint = %v", got) + } + + if err := c.AddAlias(uuid, "bafkreiaaa"); err != nil { + t.Fatalf("AddAlias: %v", err) + } + got := mustGetMetadata(t, c, uuid) + if got["cids/bafkreiaaa"] != "true" { + t.Errorf("alias key-set member missing from read: %v", got) + } + + // And the CID enumeration, which now reads the index instead of + // scanning, must see it. + cids, err := c.cidsForRootLocked(context.Background(), uuid) + if err != nil { + t.Fatalf("cidsForRootLocked: %v", err) + } + if !reflect.DeepEqual(cids, []string{"bafkreiaaa"}) { + t.Errorf("cidsForRoot = %v, want [bafkreiaaa]", cids) + } + }) +} + +// The index lives under the record's own prefix, so the "is this record empty?" +// probe would otherwise always find it and the root could never be de-indexed. +func TestFieldIndex_IndexRemoveIfEmptyIgnoresTheIndexItself(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + if err := c.SetProperty("root4", "title", "x"); err != nil { + t.Fatalf("SetProperty: %v", err) + } + + removed, err := c.IndexRemoveIfEmpty("root4") + if err != nil { + t.Fatalf("IndexRemoveIfEmpty: %v", err) + } + if removed { + t.Error("de-indexed a record that still has a field") + } + + if err := c.DeleteProperty("root4", "title"); err != nil { + t.Fatalf("DeleteProperty: %v", err) + } + removed, err = c.IndexRemoveIfEmpty("root4") + if err != nil { + t.Fatalf("IndexRemoveIfEmpty: %v", err) + } + if !removed { + t.Fatal("record with no fields left was not de-indexed — the field index masked emptiness") + } + // ...and the now-empty index key must not be left orphaned. + n, err := c.client.Exists(context.Background(), c.buildFieldsKey("root4")).Result() + if err != nil { + t.Fatalf("exists: %v", err) + } + if n != 0 { + t.Error("orphaned field index left behind after de-indexing") + } + }) +} + +// The same probe must still work for a record that predates the index. +func TestFieldIndex_IndexRemoveIfEmptyHandlesUnindexedRecord(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + ctx := context.Background() + if err := c.client.Set(ctx, c.buildKeyPrefix("legacy2")+"title", "x", 0).Err(); err != nil { + t.Fatalf("seed: %v", err) + } + if err := c.IndexAdd("legacy2"); err != nil { + t.Fatalf("IndexAdd: %v", err) + } + removed, err := c.IndexRemoveIfEmpty("legacy2") + if err != nil { + t.Fatalf("IndexRemoveIfEmpty: %v", err) + } + if removed { + t.Error("de-indexed an un-indexed record that still has a field") + } + }) +} + +// The reserved name is bookkeeping. It must never reach an API caller as a +// metadata field, and an inbound payload must never be able to overwrite the +// SET with a string — that would break every subsequent read of the record. +func TestFieldIndex_ReservedNameIsNeverMetadata(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + err := c.SetMetadataFlat("root5", map[string]string{ + "title": "x", + FieldsField: "malicious", + }) + if err != nil { + t.Fatalf("SetMetadataFlat: %v", err) + } + + got := mustGetMetadata(t, c, "root5") + if _, present := got[FieldsField]; present { + t.Errorf("%s surfaced as a metadata field: %v", FieldsField, got) + } + if !reflect.DeepEqual(got, map[string]string{"title": "x"}) { + t.Errorf("read = %v, want just title", got) + } + + // The index must still be a SET — i.e. not clobbered by the payload. + if got := fieldNames(t, c, "root5"); !reflect.DeepEqual(got, []string{"title"}) { + t.Errorf("index = %v, want [title]", got) + } + + if err := c.SetProperty("root5", FieldsField, "malicious"); err == nil { + t.Error("SetProperty accepted the reserved field name") + } + }) +} + +// Deleting a record must take its index with it, or a recreated root would +// inherit phantom field names. +func TestFieldIndex_DeleteMetadataRemovesTheIndex(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + if err := c.SetMetadataFlat("root6", map[string]string{"title": "x", "year": "2002"}); err != nil { + t.Fatalf("SetMetadataFlat: %v", err) + } + if _, err := c.DeleteMetadata("root6"); err != nil { + t.Fatalf("DeleteMetadata: %v", err) + } + if got := fieldNames(t, c, "root6"); len(got) != 0 { + t.Errorf("index survived DeleteMetadata: %v", got) + } + got, err := c.GetMetadataFlat("root6") + if err != nil { + t.Fatalf("GetMetadataFlat: %v", err) + } + if got != nil { + t.Errorf("deleted record still reads back: %v", got) + } + }) +} + +// A stale index entry (name present, key gone) is the benign direction: the +// MGET reads nil and the field is skipped rather than erroring. +func TestFieldIndex_StaleEntryIsHarmless(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + ctx := context.Background() + if err := c.SetMetadataFlat("root7", map[string]string{"title": "x"}); err != nil { + t.Fatalf("SetMetadataFlat: %v", err) + } + // Name a field that has no key behind it. + if err := c.client.SAdd(ctx, c.buildFieldsKey("root7"), "ghost").Err(); err != nil { + t.Fatalf("sadd ghost: %v", err) + } + + got := mustGetMetadata(t, c, "root7") + if !reflect.DeepEqual(got, map[string]string{"title": "x"}) { + t.Errorf("read = %v, want just title", 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. +func TestFieldIndex_SearchIndexTuplesSkipTheIndexKeys(t *testing.T) { + forEachPrefix(t, func(t *testing.T, c *Client) { + if err := c.SetMetadataFlat("root8", map[string]string{ + "title": "Naruto Shippuden", + "fileName": "naruto.mkv", + }); err != nil { + t.Fatalf("SetMetadataFlat: %v", err) + } + + tuples, err := c.GetSearchIndexTuples() + if err != nil { + t.Fatalf("GetSearchIndexTuples: %v", err) + } + var found bool + for _, tu := range tuples { + if tu.HashID != "root8" { + continue + } + found = true + if _, present := tu.Fields[FieldsField]; present { + t.Errorf("%s leaked into the search tuple: %v", FieldsField, tu.Fields) + } + } + if !found { + t.Fatal("record missing from search tuples") + } + }) +}