From a6374225fb7ae465bafbcaf22afbbe71624403cf Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 15 Feb 2026 22:38:28 +0100 Subject: [PATCH] log stream improvement --- docs/metadata-storage-structure.md | 216 +++++++++++++++++++++++------ internal/events/meta_publisher.go | 16 ++- internal/storage/client.go | 13 +- internal/watcher/dispatcher.go | 6 +- 4 files changed, 197 insertions(+), 54 deletions(-) diff --git a/docs/metadata-storage-structure.md b/docs/metadata-storage-structure.md index 766fba9..0f68bfa 100644 --- a/docs/metadata-storage-structure.md +++ b/docs/metadata-storage-structure.md @@ -4,41 +4,50 @@ This document describes how MetaMesh stores file metadata in Redis, the key stru ## Overview -MetaMesh uses Redis as its primary metadata storage, storing file metadata as **Redis Hashes** with flattened key-value pairs. This design enables: +MetaMesh uses Redis as its primary metadata storage, storing file metadata as **flat string keys**. Each property is stored as a separate Redis key rather than using Redis Hashes. This design enables: -- Efficient nested object storage without JSON serialization overhead -- Individual field updates without full document rewrites -- Cross-service access via Redis pub/sub notifications +- **Redis keyspace notifications** for field-level change detection +- **Pattern-based subscriptions** (e.g., `file:*/tmdb/*` for TMDB updates) +- **meta:events stream** integration for real-time updates across services +- Individual field updates without affecting other properties - Fast VFS lookups using content-based hash IDs ## Key Structure -### Hash-Based Storage +### Flat Key Storage -Each file's metadata is stored as a Redis Hash: +Each file property is stored as a separate Redis STRING key: ``` -Hash Key: file:{hashId} -Hash Fields: property/path (e.g., "fileinfo/duration", "titles/eng") -Hash Values: String values +Key Pattern: file:{hashId}/{property/path} +Key Type: STRING +Value: String value ``` **Example Redis commands:** ```bash -# Store file metadata -HMSET file:bafkr4ih5kapbjzq... \ - "cid_midhash256" "bafkr4ih5kapbjzq..." \ - "filePath" "/files/watch/movie.mkv" \ - "fileinfo/duration" "7200.5" \ - "fileinfo/formatName" "matroska,webm" +# Store file metadata (each property is a separate key) +SET "file:bafkr4ih5kapbjzq.../cid_midhash256" "bafkr4ih5kapbjzq..." +SET "file:bafkr4ih5kapbjzq.../filePath" "/files/watch/movie.mkv" +SET "file:bafkr4ih5kapbjzq.../fileinfo/duration" "7200.5" +SET "file:bafkr4ih5kapbjzq.../fileinfo/formatName" "matroska,webm" -# Retrieve all metadata -HGETALL file:bafkr4ih5kapbjzq... +# Retrieve single field +GET "file:bafkr4ih5kapbjzq.../fileinfo/duration" -# Get single field -HGET file:bafkr4ih5kapbjzq... "fileinfo/duration" +# Retrieve all metadata for a file (scan by prefix) +SCAN 0 MATCH "file:bafkr4ih5kapbjzq.../*" COUNT 1000 ``` +### Why Flat Keys? + +The flat key structure (vs. Redis Hashes) enables: + +1. **Keyspace Notifications**: Redis publishes events when individual keys change +2. **Pattern Subscriptions**: Services can subscribe to `__keyspace@0__:file:*/tmdb/*` +3. **Selective Updates**: Update one field without touching others +4. **Stream Integration**: The `meta:events` stream captures field-level changes + ### File Index A Redis Set tracks all known file hash IDs: @@ -51,6 +60,66 @@ Value: {hashId1, hashId2, hashId3, ...} This enables fast bulk queries like "get all files" without scanning keys. +```bash +# Get all hash IDs +SMEMBERS "file:__index__" + +# Count total files +SCARD "file:__index__" + +# Check if a file exists +SISMEMBER "file:__index__" "bafkr4ih5kapbjzq..." +``` + +### Keyspace Notifications + +Redis keyspace notifications are enabled to detect metadata changes. This requires the `notify-keyspace-events` configuration: + +```bash +# Required Redis config (set by meta-core) +CONFIG SET notify-keyspace-events "K$" + +# K = Keyspace events (published on __keyspace@__:) +# $ = String commands (SET, etc.) +``` + +Services can subscribe to keyspace events for real-time updates: + +```bash +# Subscribe to all file key changes +PSUBSCRIBE '__keyspace@0__:file:*' + +# Subscribe to specific field pattern (e.g., TMDB updates) +PSUBSCRIBE '__keyspace@0__:file:*/tmdb/*' +``` + +### meta:events Stream + +The `meta:events` stream provides reliable event delivery for file system events from meta-core: + +``` +Stream Key: meta:events +Entry Fields: + - type: add | change | delete | rename | reset + - path: File path relative to /files + - midhash256: Content hash (for add/change events) + - oldPath: Previous path (for rename events) + - timestamp: Event timestamp +``` + +Consumer groups enable reliable processing with acknowledgment: + +```bash +# Create consumer group +XGROUP CREATE meta:events mygroup 0 MKSTREAM + +# Read new events (blocking) +XREADGROUP GROUP mygroup myconsumer BLOCK 5000 STREAMS meta:events > + +# Acknowledge processed events +XACK meta:events mygroup +``` + ## Primary Hash ID The primary identifier for each file is `cid_midhash256` - an IPFS-compatible CID generated from: @@ -151,7 +220,7 @@ File Discovered └─ full-hash → cid_sha2-256, cid_sha1 ↓ [Plugin Callbacks] - └─ Merge metadata into Redis via HMSET + └─ Store metadata via flat keys (SET per property) ``` ## Core Metadata Types @@ -275,19 +344,47 @@ fileinfo/streamdetails/embeddedimage/0/type = "cover" - `language` - ISO 639-2 code - `format` - subrip, ass, pgs, vobsub -## Redis Pub/Sub Notifications +## Real-Time Notifications -MetaMesh uses Redis pub/sub to notify other services of metadata changes. +MetaMesh provides multiple mechanisms for real-time metadata change notifications: -### Channels +### 1. meta:events Stream (Recommended) + +The `meta:events` Redis stream provides reliable, ordered event delivery with consumer groups: + +```bash +# File system events from meta-core watcher +XREAD STREAMS meta:events 0 + +# Event types: add, change, delete, rename, reset +``` + +Benefits: +- **Reliable delivery** via consumer groups and acknowledgment +- **Replay capability** for catching up after restarts +- **Ordered processing** with stream IDs + +### 2. Keyspace Notifications + +For field-level change detection, subscribe to Redis keyspace notifications: + +```bash +# All file metadata changes +PSUBSCRIBE '__keyspace@0__:file:*' + +# Specific plugin updates (e.g., TMDB enrichment) +PSUBSCRIBE '__keyspace@0__:file:*/tmdb/*' +``` + +### 3. Pub/Sub Channels (Legacy) + +Batch notification channels are still supported: ``` meta-sort:file:batch # Batch file updates (every 5 seconds) meta-sort:scan:reset # Full rescan notification ``` -### Batch Update Message Format - ```json { "timestamp": 1704067200000, @@ -299,17 +396,23 @@ meta-sort:scan:reset # Full rescan notification } ``` -This enables meta-fuse to update its VFS in near real-time without polling. - ## Complete Example Here's a complete metadata example for a movie file: ```bash -# Query all metadata -redis-cli HGETALL file:bafkr4ih5kapbjzq... +# Query all keys for a file +redis-cli --scan --pattern 'file:bafkr4ih5kapbjzq.../*' -# Result (reconstructed as nested object) +# Example flat keys stored in Redis: +# file:bafkr4ih5kapbjzq.../cid_midhash256 = "bafkr4ih5kapbjzq..." +# file:bafkr4ih5kapbjzq.../filePath = "/files/watch/movies/Inception.2010.1080p.BluRay.x264.mkv" +# file:bafkr4ih5kapbjzq.../fileType = "video" +# file:bafkr4ih5kapbjzq.../fileinfo/duration = "8878.5" +# file:bafkr4ih5kapbjzq.../titles/eng = "Inception" +# ... + +# When reconstructed by the API (nested object representation) { # Content identification "cid_midhash256": "bafkr4ih5kapbjzq...", @@ -393,18 +496,49 @@ redis-cli HGETALL file:bafkr4ih5kapbjzq... ## Debugging ```bash -# List all file keys -docker exec meta-sort-dev redis-cli KEYS 'file:*' | head -20 - -# Get all metadata for a file -docker exec meta-sort-dev redis-cli HGETALL 'file:bafkr4ih5...' - -# Get specific field -docker exec meta-sort-dev redis-cli HGET 'file:bafkr4ih5...' 'fileinfo/duration' +# List all file hash IDs +docker exec meta-core-dev redis-cli SMEMBERS 'file:__index__' # Count total files -docker exec meta-sort-dev redis-cli SCARD 'file:__index__' +docker exec meta-core-dev redis-cli SCARD 'file:__index__' -# Monitor metadata changes in real-time -docker exec meta-sort-dev redis-cli SUBSCRIBE 'meta-sort:file:batch' +# Get a sample hash ID +docker exec meta-core-dev redis-cli SRANDMEMBER 'file:__index__' + +# List all keys for a specific file (flat key scan) +docker exec meta-core-dev redis-cli --scan --pattern 'file:bafkr4ih5.../*' + +# Get specific field (flat key) +docker exec meta-core-dev redis-cli GET 'file:bafkr4ih5.../fileinfo/duration' + +# Get filePath for a file +docker exec meta-core-dev redis-cli GET 'file:bafkr4ih5.../filePath' + +# Verify keys are STRING type (not HASH) +docker exec meta-core-dev redis-cli TYPE 'file:bafkr4ih5.../filePath' +# Should return: string + +# Monitor metadata changes via stream +docker exec meta-core-dev redis-cli XREAD STREAMS 'meta:events' 0 + +# Monitor keyspace notifications (requires notify-keyspace-events enabled) +docker exec meta-core-dev redis-cli PSUBSCRIBE '__keyspace@0__:file:*' ``` + +## meta:events Stream + +The `meta:events` stream captures metadata change events for real-time notifications: + +```bash +# View recent events +docker exec meta-core-dev redis-cli XRANGE 'meta:events' - + COUNT 10 + +# Read events from a consumer group +docker exec meta-core-dev redis-cli XREADGROUP GROUP mygroup myconsumer STREAMS 'meta:events' > +``` + +Stream events contain: +- `type`: Event type (add, change, delete, rename, reset) +- `path`: File path +- `midhash256`: Content hash (for add/change) +- `timestamp`: Event timestamp diff --git a/internal/events/meta_publisher.go b/internal/events/meta_publisher.go index 1f05a38..68da340 100644 --- a/internal/events/meta_publisher.go +++ b/internal/events/meta_publisher.go @@ -14,8 +14,6 @@ import ( const ( // MetaEventsStream is the Redis stream name for metadata events MetaEventsStream = "meta:events" - // StreamMaxLen is the maximum stream length (approximate) - StreamMaxLen = 10000 ) // MetaPublisher subscribes to Redis keyspace notifications @@ -49,6 +47,16 @@ func (p *MetaPublisher) Start() error { p.running = true p.mu.Unlock() + // Clear the stream on startup - consumers will re-bootstrap from Redis keys + // This ensures the stream only contains events from the current session + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := p.client.Del(ctx, MetaEventsStream).Err(); err != nil { + log.Printf("[MetaPublisher] Warning: failed to clear stream on startup: %v", err) + } else { + log.Printf("[MetaPublisher] Cleared %s stream for fresh start", MetaEventsStream) + } + p.wg.Add(1) go p.subscribeLoop() @@ -125,11 +133,9 @@ func (p *MetaPublisher) publishEvent(channel, operation string) { "ts": fmt.Sprintf("%d", time.Now().UnixMilli()), } - // Add to stream + // Add to stream (no MaxLen - stream is cleared on meta-core restart) _, err := p.client.XAdd(p.ctx, &redis.XAddArgs{ Stream: MetaEventsStream, - MaxLen: StreamMaxLen, - Approx: true, Values: fields, }).Result() diff --git a/internal/storage/client.go b/internal/storage/client.go index 3b08780..41fac51 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -633,7 +633,8 @@ func (c *Client) GetPrefix() string { } // XAdd publishes an event to a Redis Stream -// Uses XADD with MAXLEN ~ for approximate trimming +// If maxLen > 0, uses XADD with MAXLEN ~ for approximate trimming +// If maxLen <= 0, no trimming is applied (stream grows unbounded) func (c *Client) XAdd(stream string, maxLen int64, fields map[string]interface{}) (string, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -645,14 +646,18 @@ func (c *Client) XAdd(stream string, maxLen int64, fields map[string]interface{} ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - // Build XADD arguments with approximate MAXLEN + // Build XADD arguments args := &redis.XAddArgs{ Stream: stream, - MaxLen: maxLen, - Approx: true, Values: fields, } + // Only set MaxLen if positive (0 means no limit) + if maxLen > 0 { + args.MaxLen = maxLen + args.Approx = true + } + id, err := c.client.XAdd(ctx, args).Result() if err != nil { return "", fmt.Errorf("xadd failed: %w", err) diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index c9bb4b0..46ab142 100644 --- a/internal/watcher/dispatcher.go +++ b/internal/watcher/dispatcher.go @@ -11,8 +11,6 @@ import ( const ( // EventsStream is the Redis Stream name for file events EventsStream = "file:events" - // StreamMaxLen is the approximate max length for the stream - StreamMaxLen = 10000 ) // Dispatcher sends file events to Redis Stream @@ -70,8 +68,8 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) { fields["watcherId"] = event.WatcherID } - // Publish to stream - _, err := client.XAdd(EventsStream, StreamMaxLen, fields) + // Publish to stream (no MaxLen - stream is cleared on reset) + _, err := client.XAdd(EventsStream, 0, fields) if err != nil { log.Printf("[Dispatcher] Failed to publish to stream: %v", err) }