file event improvement using midhash

This commit is contained in:
Pierre
2026-02-08 17:00:54 +01:00
parent 704011fe09
commit 20e908f1b6
9 changed files with 94 additions and 56 deletions
-1
View File
@@ -508,7 +508,6 @@ services:
| `META_CORE_PATH` | `/meta-core` | Path to META_CORE_VOLUME (infrastructure) | | `META_CORE_PATH` | `/meta-core` | Path to META_CORE_VOLUME (infrastructure) |
| `FILES_PATH` | `/files` | Path to FILES_VOLUME (shared media) | | `FILES_PATH` | `/files` | Path to FILES_VOLUME (shared media) |
| `REDIS_URL` | Auto-discover | Redis connection (from leader lock file) | | `REDIS_URL` | Auto-discover | Redis connection (from leader lock file) |
| `WATCH_FOLDER_LIST` | `$FILES_PATH` | Comma-separated folders to watch for media |
| `SERVICE_NAME` | Container hostname | Unique service identifier | | `SERVICE_NAME` | Container hostname | Unique service identifier |
| `BASE_URL` | `http://localhost:8180` | External URL for this service | | `BASE_URL` | `http://localhost:8180` | External URL for this service |
+1 -8
View File
@@ -2,14 +2,7 @@
# Copy this file to .env and adjust values as needed # Copy this file to .env and adjust values as needed
# ============================================================================= # =============================================================================
# Required Settings # Volume Paths
# =============================================================================
# Comma-separated list of folders to watch for media files (REQUIRED)
WATCH_FOLDER_LIST=/files
# =============================================================================
# Volume Paths (New Architecture)
# ============================================================================= # =============================================================================
# Path to META_CORE_VOLUME - infrastructure volume for KV DB, locks, services # Path to META_CORE_VOLUME - infrastructure volume for KV DB, locks, services
@@ -12,8 +12,8 @@ import type { RedisKVClient, StreamMessage } from '../kv/RedisClient.js';
// Pipeline interface (minimal, matching StreamingPipeline) // Pipeline interface (minimal, matching StreamingPipeline)
interface Pipeline { interface Pipeline {
handleFileAdded(filePath: string): void; handleFileAdded(filePath: string, midhash256?: string): void;
handleFileChanged(filePath: string): void; handleFileChanged(filePath: string, midhash256?: string): void;
handleFileDeleted(filePath: string): void; handleFileDeleted(filePath: string): void;
} }
@@ -116,11 +116,11 @@ export class FileEventConsumer {
switch (message.type) { switch (message.type) {
case 'add': case 'add':
this.pipeline.handleFileAdded(absolutePath); this.pipeline.handleFileAdded(absolutePath, message.midhash256);
break; break;
case 'change': case 'change':
this.pipeline.handleFileChanged(absolutePath); this.pipeline.handleFileChanged(absolutePath, message.midhash256);
break; break;
case 'delete': case 'delete':
@@ -132,7 +132,7 @@ export class FileEventConsumer {
if (message.oldPath) { if (message.oldPath) {
this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath)); this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath));
} }
this.pipeline.handleFileAdded(absolutePath); this.pipeline.handleFileAdded(absolutePath, message.midhash256);
break; break;
default: default:
@@ -35,7 +35,7 @@ export interface StreamMessage {
type: 'add' | 'change' | 'delete' | 'rename'; type: 'add' | 'change' | 'delete' | 'rename';
path: string; path: string;
size?: string; size?: string;
partialHash?: string; midhash256?: string; // midhash256 CID computed by meta-core
oldPath?: string; oldPath?: string;
timestamp: string; timestamp: string;
} }
@@ -606,7 +606,7 @@ export class RedisKVClient implements IKVClient {
type: fieldMap.type as StreamMessage['type'], type: fieldMap.type as StreamMessage['type'],
path: fieldMap.path, path: fieldMap.path,
size: fieldMap.size, size: fieldMap.size,
partialHash: fieldMap.partialHash, midhash256: fieldMap.midhash256,
oldPath: fieldMap.oldPath, oldPath: fieldMap.oldPath,
timestamp: fieldMap.timestamp || '0', timestamp: fieldMap.timestamp || '0',
}; };
@@ -11,8 +11,12 @@ export interface FileAnalyzerInterface {
/** /**
* Light processing phase - fast metadata extraction + midhash256 * Light processing phase - fast metadata extraction + midhash256
* File becomes accessible in VFS with permanent ID after this phase * File becomes accessible in VFS with permanent ID after this phase
* @param filePath - Path to the file
* @param current - Current file number
* @param queueSize - Total queue size
* @param midhash256 - Pre-computed midhash256 from meta-core (if available)
*/ */
processLightPhase(filePath: string, current: number, queueSize: number):Promise<void>; processLightPhase(filePath: string, current: number, queueSize: number, midhash256?: string):Promise<void>;
/** /**
* Hash processing phase - background hash computation * Hash processing phase - background hash computation
@@ -1,6 +1,6 @@
import {HashIndexManager} from "@metazla/meta-hash"; import {HashIndexManager} from "@metazla/meta-hash";
import {config} from "../../config/EnvConfig.js"; import {config} from "../../config/EnvConfig.js";
import {stat} from "fs/promises"; import * as webdav from '../../webdav/WebdavClient.js';
import {targetHash} from "../../config/TargetHash.js"; import {targetHash} from "../../config/TargetHash.js";
import {HashMeta} from "@metazla/meta-interface"; import {HashMeta} from "@metazla/meta-interface";
import {FileAnalyzerInterface} from "./FileAnalyzerInterface.js"; import {FileAnalyzerInterface} from "./FileAnalyzerInterface.js";
@@ -50,7 +50,10 @@ export class FileProcessor implements FileAnalyzerInterface{
return; return;
} }
const stats = await stat(filePath); const stats = await webdav.stat(filePath);
if (!stats || !stats.exists) {
throw new Error(`File not found via WebDAV: ${filePath}`);
}
let indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); let indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());
let hash: HashMeta = {}; let hash: HashMeta = {};
if(indexLine) { if(indexLine) {
@@ -2,7 +2,7 @@ import {Piscina} from "piscina";
import {Input, Output} from "./worker.js"; import {Input, Output} from "./worker.js";
import {HashIndexManager} from "@metazla/meta-hash"; import {HashIndexManager} from "@metazla/meta-hash";
import {config} from "../../config/EnvConfig.js"; import {config} from "../../config/EnvConfig.js";
import {stat} from "fs/promises"; import * as webdav from '../../webdav/WebdavClient.js';
// Note: Full hashes are now computed by the full-hash container plugin // Note: Full hashes are now computed by the full-hash container plugin
import {HashMeta} from "@metazla/meta-interface"; import {HashMeta} from "@metazla/meta-interface";
import {FileAnalyzerInterface} from "./FileAnalyzerInterface.js"; import {FileAnalyzerInterface} from "./FileAnalyzerInterface.js";
@@ -160,7 +160,10 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
this.unifiedStateManager.startLightProcessing(filePath); this.unifiedStateManager.startLightProcessing(filePath);
} }
const stats = await stat(filePath); const stats = await webdav.stat(filePath);
if (!stats || !stats.exists) {
throw new Error(`File not found via WebDAV: ${filePath}`);
}
// STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files) // STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files)
let midHash256: string | undefined; let midHash256: string | undefined;
@@ -325,44 +328,65 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
/** /**
* Light Processing Phase - Fast processing to make file accessible in VFS * Light Processing Phase - Fast processing to make file accessible in VFS
* - Compute midhash256 (< 1s) * - Use midhash256 from meta-core (or compute locally if not provided)
* - Extract metadata (filename, FFmpeg analysis) * - Extract metadata (filename, FFmpeg analysis)
* - Store in KV * - Store in KV
* - File becomes accessible in VFS with midhash256 ID * - File becomes accessible in VFS with midhash256 ID
* *
* @param filePath - Path to the file
* @param current - Current file number (for progress reporting)
* @param queueSize - Total queue size (for progress reporting)
* @param providedMidHash256 - Pre-computed midhash256 from meta-core (if available)
*
* @deprecated Use TaskScheduler-based processing via WatchedFileProcessor.queueFile() instead. * @deprecated Use TaskScheduler-based processing via WatchedFileProcessor.queueFile() instead.
* This method is kept for backward compatibility with StreamingPipeline. * This method is kept for backward compatibility with StreamingPipeline.
*/ */
async processLightPhase(filePath: string, current: number, queueSize: number) { async processLightPhase(filePath: string, current: number, queueSize: number, providedMidHash256?: string) {
try { try {
// STATE: Start LIGHT processing // STATE: Start LIGHT processing
if (this.unifiedStateManager) { if (this.unifiedStateManager) {
this.unifiedStateManager.startLightProcessing(filePath); this.unifiedStateManager.startLightProcessing(filePath);
} }
const stats = await stat(filePath); const stats = await webdav.stat(filePath);
if (!stats || !stats.exists) {
throw new Error(`File not found via WebDAV: ${filePath}`);
}
// STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files) // STEP 1: Use midhash256 from meta-core if provided, otherwise check cache/compute locally
let midHash256: string | undefined; let midHash256: string | undefined;
const indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());
if (indexLine && indexLine.cid_midhash256) { if (providedMidHash256) {
// Cache hit! Use cached midhash256 (no network I/O needed) // midhash256 provided by meta-core - use directly (no file I/O needed)
midHash256 = indexLine.cid_midhash256; midHash256 = providedMidHash256;
performanceMetrics.recordCacheHit('midhash256'); performanceMetrics.recordCacheHit('midhash256');
} else {
// Cache miss - compute midhash256 (< 1 second, but requires network I/O on network drives)
const computeStart = performance.now();
const { computeMidHash256 } = await import('@metazla/meta-hash');
midHash256 = await computeMidHash256(filePath);
const computeTime = Math.ceil(performance.now() - computeStart);
// Record midhash256 computation time (excludes cache hits) // Store in local cache for future lookups
performanceMetrics.recordHashComputation('cid_midhash256', computeTime);
performanceMetrics.recordCacheMiss('midhash256');
// Store in cache for future lookups
this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 }); this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 });
} else {
// Fallback: Check cache or compute locally (legacy path)
const indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());
if (indexLine && indexLine.cid_midhash256) {
// Cache hit! Use cached midhash256 (no network I/O needed)
midHash256 = indexLine.cid_midhash256;
performanceMetrics.recordCacheHit('midhash256');
} else {
// Cache miss - compute midhash256 locally (requires file access)
// Note: This path should not be hit in normal operation since meta-core provides midhash256
console.warn(`[FileProcessor] Computing midhash256 locally for ${filePath} - meta-core should provide this`);
const computeStart = performance.now();
const { computeMidHash256 } = await import('@metazla/meta-hash');
midHash256 = await computeMidHash256(filePath);
const computeTime = Math.ceil(performance.now() - computeStart);
// Record midhash256 computation time (excludes cache hits)
performanceMetrics.recordHashComputation('cid_midhash256', computeTime);
performanceMetrics.recordCacheMiss('midhash256');
// Store in cache for future lookups
this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 });
}
} }
// Initialize metadata with midhash256 // Initialize metadata with midhash256
@@ -451,7 +475,10 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
this.unifiedStateManager.startHashProcessing(filePath); this.unifiedStateManager.startHashProcessing(filePath);
} }
const stats = await stat(filePath); const stats = await webdav.stat(filePath);
if (!stats || !stats.exists) {
throw new Error(`File not found via WebDAV: ${filePath}`);
}
const metadata = this.database.get(filePath); const metadata = this.database.get(filePath);
if (!metadata) { if (!metadata) {
throw new Error('Metadata not found - light processing may have failed'); throw new Error('Metadata not found - light processing may have failed');
@@ -2,7 +2,7 @@ import {CID_ALGORITHM_NAMES, HashComputerFile} from "@metazla/meta-hash";
import {HashIndexManager} from "@metazla/meta-hash"; import {HashIndexManager} from "@metazla/meta-hash";
import {config} from "../../config/EnvConfig.js"; import {config} from "../../config/EnvConfig.js";
import {targetHash, targetHashForIndex} from "../../config/TargetHash.js"; import {targetHash, targetHashForIndex} from "../../config/TargetHash.js";
import {stat} from "fs/promises"; import * as webdav from '../../webdav/WebdavClient.js';
export const globalHashIndex = new HashIndexManager(config.INDEX_FOLDER_PATH, targetHashForIndex); export const globalHashIndex = new HashIndexManager(config.INDEX_FOLDER_PATH, targetHashForIndex);
@@ -12,7 +12,10 @@ class SimpleFileToCid{
async getcid(filePath: string): Promise<string | null> { async getcid(filePath: string): Promise<string | null> {
try { try {
const stats = await stat(filePath); const stats = await webdav.stat(filePath);
if (!stats || !stats.exists) {
return null;
}
let indexLine = globalHashIndex.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); let indexLine = globalHashIndex.getCidForFile(filePath, stats.size, stats.mtime.toISOString());
if(indexLine) { if(indexLine) {
return indexLine[CID_ALGORITHM_NAMES.sha256]; return indexLine[CID_ALGORITHM_NAMES.sha256];
@@ -91,8 +91,10 @@ export class StreamingPipeline {
* - Optional MIME validation (reads file header) * - Optional MIME validation (reads file header)
* - Files already marked as discovered in start() * - Files already marked as discovered in start()
* - After validation passes, file is removed from discovered (tracked by PQueue pending) * - After validation passes, file is removed from discovered (tracked by PQueue pending)
* @param filePath - Path to the file
* @param midhash256 - Pre-computed midhash256 from meta-core (if available)
*/ */
private async validateFile(filePath: string): Promise<void> { private async validateFile(filePath: string, midhash256?: string): Promise<void> {
try { try {
// Extension check (instant, no I/O) // Extension check (instant, no I/O)
const ext = path.extname(filePath).toLowerCase(); const ext = path.extname(filePath).toLowerCase();
@@ -113,7 +115,8 @@ export class StreamingPipeline {
this.config.stateManager.removeFile(filePath); this.config.stateManager.removeFile(filePath);
// Stage 2: Fast Queue - metadata + midhash256 (fire and forget) // Stage 2: Fast Queue - metadata + midhash256 (fire and forget)
this.fastQueue.add(() => this.processLightPhase(filePath)) // Pass midhash256 from meta-core if available
this.fastQueue.add(() => this.processLightPhase(filePath, midhash256))
.catch(err => console.error(`[Pipeline] Fast queue error for ${filePath}:`, err.message)); .catch(err => console.error(`[Pipeline] Fast queue error for ${filePath}:`, err.message));
} catch (error: any) { } catch (error: any) {
console.error(`[Pipeline] Validation failed for ${filePath}:`, error.message); console.error(`[Pipeline] Validation failed for ${filePath}:`, error.message);
@@ -125,17 +128,19 @@ export class StreamingPipeline {
/** /**
* Stage 2: Fast Queue Phase * Stage 2: Fast Queue Phase
* - Extract metadata (filename parsing, FFmpeg analysis) * - Extract metadata (filename parsing, FFmpeg analysis)
* - Compute midhash256 (< 1s) * - Use midhash256 from meta-core (or compute locally if not provided)
* - Store to KV (file becomes accessible in VFS) * - Store to KV (file becomes accessible in VFS)
* - Queue for background processing * - Queue for background processing
* @param filePath - Path to the file
* @param midhash256 - Pre-computed midhash256 from meta-core (if available)
*/ */
private async processLightPhase(filePath: string): Promise<void> { private async processLightPhase(filePath: string, midhash256?: string): Promise<void> {
try { try {
// Call light processing // Call light processing with midhash256 from meta-core
const current = this.fastProcessedCount + 1; const current = this.fastProcessedCount + 1;
const queueSize = this.discoveredCount; const queueSize = this.discoveredCount;
await this.config.fileProcessor.processLightPhase(filePath, current, queueSize); await this.config.fileProcessor.processLightPhase(filePath, current, queueSize, midhash256);
// After light processing, add to VFS // After light processing, add to VFS
const metadata = this.config.fileProcessor.getDatabase().get(filePath); const metadata = this.config.fileProcessor.getDatabase().get(filePath);
@@ -231,23 +236,27 @@ export class StreamingPipeline {
} }
/** /**
* Handle file added event (from chokidar) * Handle file added event (from FileEventConsumer via meta-core)
* @param filePath - Path to the file
* @param midhash256 - Pre-computed midhash256 from meta-core (if available)
*/ */
async handleFileAdded(filePath: string): Promise<void> { async handleFileAdded(filePath: string, midhash256?: string): Promise<void> {
// Process through validation stage // Process through validation stage, passing midhash256
this.validationQueue.add(() => this.validateFile(filePath)) this.validationQueue.add(() => this.validateFile(filePath, midhash256))
.catch(err => console.error(`[Pipeline] Error processing added file ${filePath}:`, err.message)); .catch(err => console.error(`[Pipeline] Error processing added file ${filePath}:`, err.message));
} }
/** /**
* Handle file changed event (from chokidar) * Handle file changed event (from FileEventConsumer via meta-core)
* @param filePath - Path to the file
* @param midhash256 - Pre-computed midhash256 from meta-core (if available)
*/ */
async handleFileChanged(filePath: string): Promise<void> { async handleFileChanged(filePath: string, midhash256?: string): Promise<void> {
// Reprocess through validation stage // Reprocess through validation stage
this.config.stateManager.removeFile(filePath); this.config.stateManager.removeFile(filePath);
this.validationQueue.add(() => this.validateFile(filePath)) this.validationQueue.add(() => this.validateFile(filePath, midhash256))
.catch(err => console.error(`[Pipeline] Error processing changed file ${filePath}:`, err.message)); .catch(err => console.error(`[Pipeline] Error processing changed file ${filePath}:`, err.message));
} }