bug fix on file dispaly, document update
This commit is contained in:
@@ -0,0 +1,455 @@
|
||||
# Streaming Architecture
|
||||
|
||||
## Overview
|
||||
|
||||
This document describes the streaming event processing system used by meta-fuse to maintain VFS state. The architecture uses Redis Streams for reliable, ordered event delivery with replay capability.
|
||||
|
||||
## Core Components
|
||||
|
||||
### StreamingStateBuilder
|
||||
|
||||
**File**: `src/vfs/StreamingStateBuilder.ts`
|
||||
|
||||
The StreamingStateBuilder is the central component for processing stream events and maintaining file metadata state.
|
||||
|
||||
```typescript
|
||||
interface StreamingStateBuilderConfig {
|
||||
/** Redis client for fetching property values */
|
||||
redisClient: RedisClient;
|
||||
/** Renaming rules config for extracting VFS-relevant properties */
|
||||
rulesConfig: RenamingConfig;
|
||||
/** VFS update callback */
|
||||
vfsCallback?: VFSUpdateCallback;
|
||||
/** FILES_VOLUME path prefix */
|
||||
filesPath?: string;
|
||||
}
|
||||
```
|
||||
|
||||
**Responsibilities**:
|
||||
- Process events from `meta:events` stream
|
||||
- Filter events by VFS-relevant properties
|
||||
- Fetch property values from Redis
|
||||
- Maintain in-memory state of all files
|
||||
- Notify VFS when files are complete or change
|
||||
|
||||
### RulesPropertyExtractor
|
||||
|
||||
**File**: `src/vfs/RulesPropertyExtractor.ts`
|
||||
|
||||
Analyzes renaming rules configuration to determine which metadata properties are needed for VFS path computation.
|
||||
|
||||
```typescript
|
||||
class RulesPropertyExtractor {
|
||||
// Extract all VFS-relevant properties from rules config
|
||||
extractVfsProperties(config: RenamingConfig): Set<string>;
|
||||
|
||||
// Get core properties that are always required
|
||||
getCoreProperties(): Set<string>;
|
||||
}
|
||||
```
|
||||
|
||||
**Core Properties** (always required):
|
||||
- `filePath` - Source file location
|
||||
- `size`, `fileSize` - File size
|
||||
- `mtime`, `ctime` - Timestamps
|
||||
- `fileName`, `extension` - Filename components
|
||||
|
||||
**Rule-Derived Properties** (extracted from templates and conditions):
|
||||
- Template variables: `{title}`, `{season:pad2}`, `{episode:pad2}`
|
||||
- Condition fields: `fileType`, `year`, `mediaType`
|
||||
|
||||
### VFSUpdateCallback
|
||||
|
||||
**File**: `src/vfs/StreamingStateBuilder.ts:32-37`
|
||||
|
||||
Interface for VFS to receive updates from StreamingStateBuilder:
|
||||
|
||||
```typescript
|
||||
interface VFSUpdateCallback {
|
||||
// Called when a property value changes
|
||||
onPropertyChange(hashId: string, property: string, value: string): void;
|
||||
|
||||
// Called when a property is deleted
|
||||
onPropertyDelete(hashId: string, property: string): void;
|
||||
|
||||
// Called when a file is completely removed (filePath deleted)
|
||||
onFileDelete(hashId: string): void;
|
||||
|
||||
// Called when a file becomes complete (has filePath) or updates
|
||||
onFileComplete(hashId: string, metadata: Record<string, string>): void;
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Event Processing Pipeline
|
||||
|
||||
### Stream Message Format
|
||||
|
||||
**File**: `src/kv/RedisClient.ts:30-43`
|
||||
|
||||
```typescript
|
||||
interface StreamMessage {
|
||||
id: string; // Stream entry ID (e.g., "1703808000000-0")
|
||||
type: 'set' | 'del'; // Operation type
|
||||
key?: string; // Redis key (e.g., "file:abc123/title")
|
||||
ts?: string; // Timestamp
|
||||
timestamp: string; // Fallback timestamp
|
||||
}
|
||||
```
|
||||
|
||||
### 5-Step Event Processing
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ Step 1: Receive Stream Message │
|
||||
│ ────────────────────────────────────────────────────────────── │
|
||||
│ {id: "1703808000-0", type: "set", key: "file:abc123/title"} │
|
||||
└───────────────────────────────┬─────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ Step 2: Parse Key │
|
||||
│ ────────────────────────────────────────────────────────────── │
|
||||
│ parseMetaEventKey("file:abc123/title") │
|
||||
│ → { hashId: "abc123", property: "title" } │
|
||||
└───────────────────────────────┬─────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ Step 3: Check VFS Relevance │
|
||||
│ ────────────────────────────────────────────────────────────── │
|
||||
│ isVfsRelevantProperty("title", vfsRelevantProps) │
|
||||
│ → true (if "title" is in renaming rules) │
|
||||
│ → false → skip, increment propertiesSkipped │
|
||||
└───────────────────────────────┬─────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ Step 4: Fetch Property Value │
|
||||
│ ────────────────────────────────────────────────────────────── │
|
||||
│ await redisClient.getProperty("abc123", "title") │
|
||||
│ → "Inception" │
|
||||
└───────────────────────────────┬─────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ Step 5: Update State & Notify VFS │
|
||||
│ ────────────────────────────────────────────────────────────── │
|
||||
│ updateProperty("abc123", "title", "Inception") │
|
||||
│ vfsCallback.onPropertyChange("abc123", "title", "Inception") │
|
||||
│ │
|
||||
│ If file just became complete (now has filePath): │
|
||||
│ vfsCallback.onFileComplete("abc123", metadata) │
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### Code Implementation
|
||||
|
||||
```typescript
|
||||
// StreamingStateBuilder.ts:163-206
|
||||
private async handlePropertySet(hashId: string, property: string): Promise<void> {
|
||||
// Step 3: Check if this property is VFS-relevant
|
||||
if (!isVfsRelevantProperty(property, this.vfsRelevantProps)) {
|
||||
this.stats.propertiesSkipped++;
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 4: Fetch the property value from Redis
|
||||
const value = await this.redisClient.getProperty(hashId, property);
|
||||
if (value === null) {
|
||||
return; // Property was set then deleted before we could fetch
|
||||
}
|
||||
|
||||
this.stats.propertiesFetched++;
|
||||
|
||||
// Check if file was complete BEFORE updating state
|
||||
const wasComplete = this.isFileComplete(hashId);
|
||||
|
||||
// Step 5: Update internal state
|
||||
this.updateProperty(hashId, property, value);
|
||||
|
||||
// Notify VFS of changes
|
||||
if (this.vfsCallback) {
|
||||
this.vfsCallback.onPropertyChange(hashId, property, value);
|
||||
|
||||
// If file just became complete (has filePath now), notify VFS
|
||||
if (!wasComplete && this.isFileComplete(hashId)) {
|
||||
this.stats.filesComplete++;
|
||||
const metadata = this.getFileMetadata(hashId);
|
||||
if (metadata) {
|
||||
this.vfsCallback.onFileComplete(hashId, metadata);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Bootstrap vs Live Phases
|
||||
|
||||
### Bootstrap Phase
|
||||
|
||||
**When**: Application startup
|
||||
|
||||
**Goal**: Rebuild complete VFS state from event history
|
||||
|
||||
```typescript
|
||||
// index.ts:158-174
|
||||
// Bootstrap: Replay meta:events stream from position 0
|
||||
const lastId = await redisClient.replayStream(
|
||||
EVENTS_STREAM,
|
||||
'0', // Start from beginning
|
||||
async (message: StreamMessage) => {
|
||||
await stateBuilder.processEvent(message);
|
||||
},
|
||||
100 // Batch size
|
||||
);
|
||||
```
|
||||
|
||||
**Characteristics**:
|
||||
- Reads from stream position `0` (all history)
|
||||
- Non-blocking batch reads (no XREADGROUP)
|
||||
- Processes all events sequentially
|
||||
- Returns last processed event ID
|
||||
|
||||
### Live Phase
|
||||
|
||||
**When**: After bootstrap completes
|
||||
|
||||
**Goal**: Process new events in real-time
|
||||
|
||||
```typescript
|
||||
// index.ts:177-187
|
||||
redisClient.startSimpleStreamConsumer(
|
||||
EVENTS_STREAM,
|
||||
lastId, // Resume from bootstrap position
|
||||
async (message: StreamMessage) => {
|
||||
await stateBuilder.processEvent(message);
|
||||
},
|
||||
5000 // 5 second block timeout
|
||||
);
|
||||
```
|
||||
|
||||
**Characteristics**:
|
||||
- Resumes from last bootstrap position
|
||||
- Uses blocking reads (XREAD BLOCK)
|
||||
- Processes events as they arrive
|
||||
- Runs until shutdown
|
||||
|
||||
---
|
||||
|
||||
## Memory Efficiency
|
||||
|
||||
### State Structure
|
||||
|
||||
```typescript
|
||||
type FileState = Map<string, string>; // property → value
|
||||
type FilesState = Map<string, FileState>; // hashId → properties
|
||||
```
|
||||
|
||||
### Memory Per File
|
||||
|
||||
| Component | Size |
|
||||
|-----------|------|
|
||||
| hashId (64 char) | ~128 bytes |
|
||||
| Map overhead | ~56 bytes |
|
||||
| 5-10 properties (avg) | ~200-400 bytes |
|
||||
| **Total per file** | **~400-600 bytes** |
|
||||
|
||||
### Comparison with Full Load
|
||||
|
||||
| Approach | Memory per File | 10K Files |
|
||||
|----------|-----------------|-----------|
|
||||
| Full HGETALL (all props) | ~2-5 KB | 20-50 MB |
|
||||
| Streaming (VFS props only) | ~500 bytes | 5 MB |
|
||||
| **Savings** | **75-90%** | **15-45 MB** |
|
||||
|
||||
### Why It Works
|
||||
|
||||
1. **Property Filtering**: Only fetch properties used in renaming rules
|
||||
2. **Lazy Loading**: Properties fetched on-demand from events
|
||||
3. **No Duplication**: Values stored once in state map
|
||||
4. **Garbage Collection**: File state removed when file deleted
|
||||
|
||||
---
|
||||
|
||||
## Property Filtering Details
|
||||
|
||||
### Extracting VFS-Relevant Properties
|
||||
|
||||
```typescript
|
||||
// RulesPropertyExtractor.ts:132-155
|
||||
extractVfsProperties(config: RenamingConfig): Set<string> {
|
||||
const properties = new Set<string>(CORE_PROPERTIES);
|
||||
|
||||
// Extract from all enabled rules
|
||||
for (const rule of config.rules) {
|
||||
if (!rule.enabled) continue;
|
||||
|
||||
// From template: {title|originalTitle}, {season:pad2}
|
||||
const templateVars = templateEngine.extractVariables(rule.template);
|
||||
properties.add(...templateVars);
|
||||
|
||||
// From conditions: { field: 'fileType', value: 'video' }
|
||||
const conditionFields = extractConditionFields(rule.conditions);
|
||||
properties.add(...conditionFields);
|
||||
}
|
||||
|
||||
return properties;
|
||||
}
|
||||
```
|
||||
|
||||
### Property Path Matching
|
||||
|
||||
Handles both exact and nested property matching:
|
||||
|
||||
```typescript
|
||||
// RulesPropertyExtractor.ts:85-111
|
||||
function isVfsRelevantProperty(property: string, vfsProperties: Set<string>): boolean {
|
||||
const normalized = normalizePropertyPath(property);
|
||||
|
||||
// Direct match
|
||||
if (vfsProperties.has(normalized)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Nested property match
|
||||
// e.g., if we track 'titles', then 'titles/eng' matches
|
||||
for (const vfsProp of vfsProperties) {
|
||||
if (normalized.startsWith(vfsProp + '.')) {
|
||||
return true;
|
||||
}
|
||||
if (vfsProp.startsWith(normalized + '.')) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Statistics and Monitoring
|
||||
|
||||
### StreamingStateBuilder Stats
|
||||
|
||||
```typescript
|
||||
// StreamingStateBuilder.ts:68-74
|
||||
private stats = {
|
||||
eventsProcessed: 0, // Total events received
|
||||
propertiesFetched: 0, // Properties fetched from Redis
|
||||
propertiesSkipped: 0, // Non-VFS-relevant properties skipped
|
||||
filesComplete: 0, // Files that became complete
|
||||
lastEventId: '0', // Last processed event ID
|
||||
};
|
||||
```
|
||||
|
||||
### Accessing Stats
|
||||
|
||||
```typescript
|
||||
const stats = stateBuilder.getStats();
|
||||
// Returns:
|
||||
{
|
||||
eventsProcessed: 15234,
|
||||
propertiesFetched: 8934,
|
||||
propertiesSkipped: 6300,
|
||||
filesComplete: 5123,
|
||||
lastEventId: "1703808000000-0",
|
||||
fileCount: 5123, // Complete files (have filePath)
|
||||
stateSize: 5234 // Total files in state
|
||||
}
|
||||
```
|
||||
|
||||
### Interpreting Stats
|
||||
|
||||
| Metric | Meaning |
|
||||
|--------|---------|
|
||||
| `eventsProcessed` | Total stream events consumed |
|
||||
| `propertiesFetched` | VFS-relevant property GET calls |
|
||||
| `propertiesSkipped` | Non-relevant events filtered |
|
||||
| `filesComplete` | Files added to VFS |
|
||||
| `fileCount` | Current files in VFS |
|
||||
| Fetch ratio | `propertiesFetched / eventsProcessed` (lower is better) |
|
||||
|
||||
---
|
||||
|
||||
## Error Handling
|
||||
|
||||
### Event Processing Errors
|
||||
|
||||
```typescript
|
||||
// StreamingStateBuilder.ts:123-136
|
||||
async processEvent(message: StreamMessage): Promise<void> {
|
||||
this.stats.eventsProcessed++;
|
||||
this.stats.lastEventId = message.id;
|
||||
|
||||
if ((message.type === 'set' || message.type === 'del') && message.key) {
|
||||
await this.processMetaEvent(message);
|
||||
return;
|
||||
}
|
||||
|
||||
// Ignore legacy event types - not supported in flat key architecture
|
||||
logger.debug(`Ignoring legacy event type: ${message.type}`);
|
||||
}
|
||||
```
|
||||
|
||||
### Property Fetch Failures
|
||||
|
||||
```typescript
|
||||
// StreamingStateBuilder.ts:171-177
|
||||
const value = await this.redisClient.getProperty(hashId, property);
|
||||
if (value === null) {
|
||||
// Property was set then deleted before we could fetch it
|
||||
// Or it's a different data type - skip silently
|
||||
return;
|
||||
}
|
||||
```
|
||||
|
||||
### Stream Consumer Recovery
|
||||
|
||||
```typescript
|
||||
// index.ts:185-187
|
||||
.catch(error => {
|
||||
logger.error('Stream consumer error:', error);
|
||||
});
|
||||
```
|
||||
|
||||
On error, the consumer loop continues retrying with brief pauses.
|
||||
|
||||
---
|
||||
|
||||
## Configuration
|
||||
|
||||
### StreamingStateBuilder Configuration
|
||||
|
||||
```typescript
|
||||
const stateBuilder = new StreamingStateBuilder({
|
||||
redisClient,
|
||||
rulesConfig: vfs.getRulesConfig(),
|
||||
vfsCallback: vfs,
|
||||
filesPath: '/files',
|
||||
});
|
||||
```
|
||||
|
||||
### Dynamic Rules Update
|
||||
|
||||
When renaming rules change:
|
||||
|
||||
```typescript
|
||||
// StreamingStateBuilder.ts:99-102
|
||||
updateRulesConfig(rulesConfig: RenamingConfig): void {
|
||||
this.vfsRelevantProps = this.propertyExtractor.extractVfsProperties(rulesConfig);
|
||||
logger.info(`Updated VFS-relevant properties: ${this.vfsRelevantProps.size}`);
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Related Documentation
|
||||
|
||||
- [VFS Rebuild Architecture](vfs-rebuild-architecture.md) - High-level architecture overview
|
||||
- [API Reference](api-reference.md) - REST API documentation
|
||||
- [Redis Key Format](../../../meta-core/docs/redis-format.md) - Flat key storage format
|
||||
Reference in New Issue
Block a user