2 Commits
Author SHA1 Message Date
Worph 021991b00d improve stability
Build and Push Docker Image / build (push) Failing after 14s
2026-09-13 12:02:27 +02:00
Worph 862f73d26d cid rank remove
Build and Push Docker Image / build (push) Failing after 13s
2026-09-09 17:58:15 +02:00
10 changed files with 435 additions and 320 deletions
@@ -11,8 +11,9 @@ import { performanceMetrics } from '../metrics/PerformanceMetrics.js';
import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js';
import type { IKVClient } from '../kv/IKVClient.js';
import type { KVManager } from '../kv/KVManager.js';
import type { ExtendedProcessingSnapshot, IKVClientWithPubSub } from '../types/ExtendedInterfaces.js';
import { getErrorMessage } from '../types/ExtendedInterfaces.js';
import type { ExtendedProcessingSnapshot } from '../types/ExtendedInterfaces.js';
import { getErrorMessage, hasFileTuples } from '../types/ExtendedInterfaces.js';
import type { IKVClientWithFileTuples } from '../types/ExtendedInterfaces.js';
import { config } from '../config/EnvConfig.js';
import * as webdav from '../webdav/WebdavClient.js';
import type { PluginManager } from '../plugin-engine/PluginManager.js';
@@ -29,6 +30,45 @@ export interface UnifiedAPIServerConfig {
enableCors?: boolean;
}
/** File totals served by /api/stats. Null when meta-core cannot supply them. */
interface FileStats {
fileCount: number | null;
totalSize: number | null;
source: 'meta-core' | 'unavailable';
}
/** Metadata reads in flight at once while a recompute builds its task list. */
const RECOMPUTE_READ_CONCURRENCY = 8;
/** Run `worker` over `items` with at most `limit` calls in flight. */
async function forEachWithConcurrency<T>(
items: readonly T[],
limit: number,
worker: (item: T) => Promise<void>
): Promise<void> {
let next = 0;
const runners = Array.from({ length: Math.min(limit, items.length) }, async () => {
while (next < items.length) {
const item = items[next++];
await worker(item);
}
});
await Promise.all(runners);
}
/** Flatten nested metadata into the `a/b/c` keys the task scheduler expects. */
function flattenMetadata(obj: Record<string, unknown>, out: Record<string, string> = {}, prefix = ''): Record<string, string> {
for (const [key, value] of Object.entries(obj)) {
const fullKey = prefix ? `${prefix}/${key}` : key;
if (value !== null && typeof value === 'object' && !Array.isArray(value)) {
flattenMetadata(value as Record<string, unknown>, out, fullKey);
} else if (value !== undefined && value !== null) {
out[fullKey] = String(value);
}
}
return out;
}
export class UnifiedAPIServer {
private app: FastifyInstance;
private unifiedStateManager: UnifiedProcessingStateManager | null = null;
@@ -44,9 +84,15 @@ export class UnifiedAPIServer {
private containerPluginScheduler: ContainerPluginScheduler | null = null;
private streamingPipeline: StreamingPipeline | null = null;
// Cache for total file size (avoid running du on every request)
private totalSizeCache: { value: number; timestamp: number } | null = null;
private static TOTAL_SIZE_CACHE_TTL_MS = 30000; // 30 seconds
// /api/stats file totals: stale-while-revalidate, one refresh at a time.
// Never walk hashIds for these — on a shared meta-core that is every gateway
// record too (82k on watch.nsl.sh), and a sweep per poll stacked unbounded.
private fileStats: { value: FileStats; fetchedAt: number } | null = null;
private fileStatsInFlight: Promise<FileStats> | null = null;
private static FILE_STATS_TTL_MS = 30000; // 30 seconds
// Plugins whose recompute is currently being queued (single-flight per plugin)
private recomputeInFlight = new Set<string>();
constructor(
config: UnifiedAPIServerConfig = {},
@@ -97,48 +143,46 @@ export class UnifiedAPIServer {
}
/**
* Get total size of processed files from Redis sizeByte field (cached)
* This is "eventually correct" - accumulates as files are processed by the file-info plugin
* File count + total size of the files meta-core holds on disk, for the
* monitor's "Total Size" card: one summary call to meta-core's
* GET /api/files/tuples (batched server-side), cached for FILE_STATS_TTL_MS.
*
* Stale-while-revalidate with a single in-flight refresh — a poll never waits
* on meta-core once a value exists, and concurrent polls share one request.
*/
private async getTotalFileSizeFromRedis(): Promise<number> {
// Check cache
const now = Date.now();
if (this.totalSizeCache && (now - this.totalSizeCache.timestamp) < UnifiedAPIServer.TOTAL_SIZE_CACHE_TTL_MS) {
return this.totalSizeCache.value;
private async getFileStats(): Promise<FileStats> {
const cached = this.fileStats;
if (cached && Date.now() - cached.fetchedAt < UnifiedAPIServer.FILE_STATS_TTL_MS) {
return cached.value;
}
// Sum sizeByte from all file entries in Redis
let totalSize = 0;
if (!this.kvClient) {
return 0;
if (!this.fileStatsInFlight) {
this.fileStatsInFlight = this.fetchFileStats()
.catch((error: unknown): FileStats => {
console.warn('[API] /api/stats: meta-core file totals unavailable:', getErrorMessage(error));
return cached?.value ?? { fileCount: null, totalSize: null, source: 'unavailable' };
})
.then((value) => {
this.fileStats = { value, fetchedAt: Date.now() };
return value;
})
.finally(() => {
this.fileStatsInFlight = null;
});
}
try {
// Get all file hash IDs
const hashIds = await this.kvClient.getAllHashIds();
// Sum up sizeByte values (batch process for efficiency)
for (const hashId of hashIds) {
try {
const sizeByte = await this.kvClient.getMetadata(hashId, 'sizeByte');
if (sizeByte !== null && sizeByte !== undefined) {
const size = typeof sizeByte === 'number' ? sizeByte : parseInt(String(sizeByte), 10);
if (!isNaN(size) && size > 0) {
totalSize += size;
}
}
} catch {
// Skip entries without sizeByte
}
}
} catch (error) {
console.warn('[API] Failed to calculate total size from Redis:', error);
return cached ? cached.value : this.fileStatsInFlight;
}
// Update cache
this.totalSizeCache = { value: totalSize, timestamp: now };
return totalSize;
private async fetchFileStats(): Promise<FileStats> {
if (!this.kvClient || !hasFileTuples(this.kvClient)) {
return { fileCount: null, totalSize: null, source: 'unavailable' };
}
const summary = await this.kvClient.getFileTuples({ summary: true });
if (!summary) {
return { fileCount: null, totalSize: null, source: 'unavailable' };
}
return { fileCount: summary.count, totalSize: summary.totalSize, source: 'meta-core' };
}
/**
@@ -524,53 +568,17 @@ export class UnifiedAPIServer {
/**
* Setup Stats API routes (Redis stats for monitor UI)
* Setup Stats API routes (file totals for the monitor UI)
*/
private setupStatsRoutes(): void {
if (!this.kvClient) {
return;
}
// Get Redis/KV stats (for monitor UI)
this.app.get('/api/stats', async (request, reply) => {
try {
const hashIds = await this.kvClient!.getAllHashIds();
const redis = (this.kvClient as IKVClientWithPubSub).getRedisClient?.() as { info: (section: string) => Promise<string> } | undefined;
let memoryUsage = 'N/A';
let memoryUsageBytes = 0;
if (redis) {
try {
const info = await redis.info('memory');
const match = info.match(/used_memory_human:(\S+)/);
const bytesMatch = info.match(/used_memory:(\d+)/);
if (match) {
memoryUsage = match[1];
}
if (bytesMatch) {
memoryUsageBytes = parseInt(bytesMatch[1], 10);
}
} catch {
// Ignore memory info errors
}
}
// Get total size from Redis sizeByte (eventually correct as files are processed)
const totalSize = await this.getTotalFileSizeFromRedis();
return {
fileCount: hashIds.length,
keyCount: hashIds.length,
totalSize,
memoryUsage,
memoryUsageBytes
};
} catch (error: any) {
return reply.status(500).send({
error: 'Failed to get stats',
details: error.message
});
}
// At most one meta-core call per FILE_STATS_TTL_MS however often the
// monitor polls — see getFileStats.
this.app.get('/api/stats', async () => {
return this.getFileStats();
});
}
@@ -886,38 +894,43 @@ export class UnifiedAPIServer {
return reply.status(400).send({ error: `Plugin '${pluginId}' is not active` });
}
if (!hasFileTuples(this.kvClient)) {
return reply.status(501).send({ error: 'Recompute needs meta-core GET /api/files/tuples' });
}
const kvClient = this.kvClient as IKVClientWithFileTuples;
if (this.recomputeInFlight.has(pluginId)) {
return reply.status(409).send({ error: `Recompute for plugin '${pluginId}' is already being queued` });
}
this.recomputeInFlight.add(pluginId);
try {
// Get all files from KV
const hashIds = await this.kvClient.getAllHashIds();
console.log(`[Recompute] Starting recompute for plugin '${pluginId}' on ${hashIds.length} files`);
// Only file-backed records can be recomputed, and meta-core lists exactly
// those in one batched call. Walking getAllHashIds() instead reads every
// gateway record too (82k on watch.nsl.sh) to keep a few thousand.
const tuples = await kvClient.getFileTuples();
if (!tuples || !tuples.files) {
return reply.status(501).send({
error: 'Recompute needs meta-core GET /api/files/tuples',
details: 'meta-core did not return a file list'
});
}
const candidates = tuples.files;
console.log(`[Recompute] Starting recompute for plugin '${pluginId}' on ${candidates.length} files`);
// Prepare file data with metadata for TaskScheduler
const files: Array<{ filePath: string; fileHash: string; kvData?: Record<string, string> }> = [];
for (const hashId of hashIds) {
const metadata = await this.kvClient.getMetadataFlat(hashId);
if (metadata && metadata.filePath) {
// Convert nested metadata to flat key-value pairs for KVStore
const kvData: Record<string, string> = {};
const flattenObj = (obj: any, prefix = '') => {
for (const [key, value] of Object.entries(obj)) {
const fullKey = prefix ? `${prefix}/${key}` : key;
if (value !== null && typeof value === 'object' && !Array.isArray(value)) {
flattenObj(value, fullKey);
} else if (value !== undefined && value !== null) {
kvData[fullKey] = String(value);
await forEachWithConcurrency(candidates, RECOMPUTE_READ_CONCURRENCY, async (tuple) => {
const metadata = await kvClient.getMetadataFlat(tuple.hashId);
if (!metadata) {
return;
}
}
};
flattenObj(metadata);
files.push({
fileHash: hashId,
filePath: metadata.filePath,
kvData
fileHash: tuple.hashId,
filePath: metadata.filePath || tuple.filePath,
kvData: flattenMetadata(metadata)
});
});
}
}
// Create and enqueue tasks with forceRecompute option
const tasks = taskScheduler.createTasksForPluginOnFiles(
@@ -935,7 +948,7 @@ export class UnifiedAPIServer {
status: 'ok',
message: `Recompute triggered for plugin '${pluginId}'`,
filesQueued: tasks.length,
totalFiles: hashIds.length
totalFiles: candidates.length
};
} catch (error: any) {
console.error(`[Recompute] Error triggering recompute for plugin '${pluginId}':`, error);
@@ -943,6 +956,8 @@ export class UnifiedAPIServer {
error: 'Failed to trigger recompute',
details: error.message
});
} finally {
this.recomputeInFlight.delete(pluginId);
}
});
}
+176
View File
@@ -8,6 +8,7 @@
import { expect } from 'chai';
import { UnifiedAPIServer } from './UnifiedAPIServer.js';
import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js';
import { performanceMetrics } from '../metrics/PerformanceMetrics.js';
import type { IKVClient } from '../kv/IKVClient.js';
import type { FastifyInstance } from 'fastify';
@@ -23,6 +24,25 @@ class MockKVClient implements IKVClient {
private data: Map<string, Map<string, any>> = new Map();
private rawStore: Map<string, string> = new Map();
// Stand-in for meta-core's GET /api/files/tuples. null = a meta-core that
// predates the endpoint.
fileTuples: Array<{ hashId: string; filePath: string; sizeByte: number; mtimeNano: number }> | null = null;
fileTuplesDelayMs = 0;
calls = { getFileTuples: 0, getAllHashIds: 0, getMetadataFlat: 0 };
async getFileTuples(opts: { summary?: boolean } = {}) {
this.calls.getFileTuples++;
if (this.fileTuplesDelayMs) {
await new Promise(resolve => setTimeout(resolve, this.fileTuplesDelayMs));
}
if (!this.fileTuples) return null;
const files = this.fileTuples;
const totalSize = files.reduce((sum, f) => sum + f.sizeByte, 0);
return opts.summary
? { count: files.length, totalSize }
: { count: files.length, totalSize, files };
}
// Basic operations
async set(key: string, value: any): Promise<void> {
this.rawStore.set(key, JSON.stringify(value));
@@ -99,6 +119,7 @@ class MockKVClient implements IKVClient {
}
async getMetadataFlat(hashId: string): Promise<any | null> {
this.calls.getMetadataFlat++;
const entry = this.data.get(hashId);
if (!entry) return null;
return Object.fromEntries(entry);
@@ -125,6 +146,7 @@ class MockKVClient implements IKVClient {
}
async getAllHashIds(): Promise<string[]> {
this.calls.getAllHashIds++;
return Array.from(this.data.keys());
}
@@ -736,3 +758,157 @@ describe('UnifiedAPIServer', function() {
});
});
});
// =============================================================================
// File totals + recompute against meta-core's GET /api/files/tuples
// =============================================================================
//
// Each test builds its own server: /api/stats caches for 30 s and recompute is
// single-flight, so shared state between tests would hide both behaviours.
describe('UnifiedAPIServer file totals and recompute', function() {
const tuple = (hashId: string, sizeByte: number) =>
({ hashId, filePath: `/files/watch/${hashId}.mkv`, sizeByte, mtimeNano: 1 });
let app: FastifyInstance;
let kv: MockKVClient;
let queued: Array<{ pluginId: string; fileHash: string }>;
beforeEach(async function() {
kv = new MockKVClient();
queued = [];
const pluginManager = { getPlugins: () => [{ id: 'ffmpeg', active: true }] };
const taskScheduler = {
createTasksForPluginOnFiles: (pluginId: string, files: Array<{ fileHash: string }>) =>
files.map(f => ({ pluginId, fileHash: f.fileHash })),
enqueueTasks: (tasks: Array<{ pluginId: string; fileHash: string }>) => { queued.push(...tasks); },
};
const server = new UnifiedAPIServer(
{ port: 0, host: 'localhost', enableCors: false },
new MockProcessingStateManager() as any,
kv as any,
4,
16,
() => ({ fast: { pending: 0, running: 0 }, background: { pending: 0, running: 0 } }),
undefined,
() => pluginManager as any,
() => taskScheduler as any
);
app = server.getApp();
await app.ready();
});
afterEach(async function() {
await app.close();
});
it('GET /api/stats sums meta-core file tuples without walking hash ids', async function() {
kv.fileTuples = [tuple('a', 100), tuple('b', 250)];
const response = await app.inject({ method: 'GET', url: '/api/stats' });
expect(response.statusCode).to.equal(200);
expect(JSON.parse(response.payload)).to.deep.equal({ fileCount: 2, totalSize: 350, source: 'meta-core' });
expect(kv.calls.getAllHashIds).to.equal(0);
expect(kv.calls.getMetadataFlat).to.equal(0);
});
it('GET /api/stats degrades to nulls on a meta-core without the endpoint', async function() {
kv.fileTuples = null;
const response = await app.inject({ method: 'GET', url: '/api/stats' });
expect(response.statusCode).to.equal(200);
expect(JSON.parse(response.payload)).to.deep.equal({ fileCount: null, totalSize: null, source: 'unavailable' });
expect(kv.calls.getAllHashIds).to.equal(0);
});
it('GET /api/stats shares one meta-core call between concurrent polls', async function() {
kv.fileTuples = [tuple('a', 100)];
kv.fileTuplesDelayMs = 50;
const responses = await Promise.all(
Array.from({ length: 10 }, () => app.inject({ method: 'GET', url: '/api/stats' }))
);
for (const response of responses) {
expect(response.statusCode).to.equal(200);
expect(JSON.parse(response.payload).totalSize).to.equal(100);
}
expect(kv.calls.getFileTuples).to.equal(1);
});
it('POST recompute returns 501 on a meta-core without the endpoint', async function() {
kv.fileTuples = null;
const response = await app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' });
expect(response.statusCode).to.equal(501);
expect(kv.calls.getAllHashIds).to.equal(0);
expect(queued).to.have.length(0);
});
it('POST recompute queues only file-backed records', async function() {
kv.seedData('a', { filePath: '/files/watch/a.mkv', fileType: 'video' });
kv.seedData('b', { filePath: '/files/watch/b.mkv', fileType: 'video' });
kv.seedData('gateway-record', { title: 'Not a file' });
kv.fileTuples = [tuple('a', 100), tuple('b', 250)];
const response = await app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' });
expect(response.statusCode).to.equal(200);
const body = JSON.parse(response.payload);
expect(body.filesQueued).to.equal(2);
expect(body.totalFiles).to.equal(2);
expect(queued.map(q => q.fileHash).sort()).to.deep.equal(['a', 'b']);
expect(kv.calls.getMetadataFlat).to.equal(2);
expect(kv.calls.getAllHashIds).to.equal(0);
});
it('POST recompute refuses a second run for the same plugin while one is queuing', async function() {
kv.seedData('a', { filePath: '/files/watch/a.mkv' });
kv.fileTuples = [tuple('a', 100)];
kv.fileTuplesDelayMs = 50;
const [first, second] = await Promise.all([
app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' }),
app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' }),
]);
expect([first.statusCode, second.statusCode].sort()).to.deep.equal([200, 409]);
expect(queued).to.have.length(1);
});
});
// =============================================================================
// File-level processing metric
// =============================================================================
describe('UnifiedProcessingStateManager file metric', function() {
// Files announced over SSE go straight to light processing; they are never
// marked discovered, which is the path that used to leave "Processed" at 0.
const runFile = (manager: UnifiedProcessingStateManager, filePath: string, error?: string) => {
manager.startLightProcessing(filePath);
manager.completeLightProcessing(filePath, 'bagacbabaeexamplehash');
manager.startHashProcessing(filePath);
manager.completeHashProcessing(filePath, 'bagacbabaeexamplehash', undefined, error);
};
it('counts a file as processed when its hash phase completes cleanly', function() {
const manager = new UnifiedProcessingStateManager();
const before = performanceMetrics.getMetrics().totalFilesProcessed;
runFile(manager, '/files/watch/metric-ok.mkv');
expect(performanceMetrics.getMetrics().totalFilesProcessed).to.equal(before + 1);
});
it('does not count a file whose hash phase failed', function() {
const manager = new UnifiedProcessingStateManager();
const before = performanceMetrics.getMetrics().totalFilesProcessed;
runFile(manager, '/files/watch/metric-failed.mkv', 'plugin failed');
expect(performanceMetrics.getMetrics().totalFilesProcessed).to.equal(before);
});
});
@@ -19,6 +19,24 @@ import {
buildFilePrefix,
} from './MetadataUtils.js';
/** One file-backed root as listed by meta-core's GET /api/files/tuples. */
export interface FileTuple {
hashId: string;
filePath: string;
sizeByte: number;
mtimeNano: number;
}
/**
* GET /api/files/tuples body. `files` is present unless the summary form was
* asked for.
*/
export interface FileTuplesResult {
count: number;
totalSize: number;
files?: FileTuple[];
}
export interface MetaCoreApiWriterConfig {
/** Base URL for meta-core's HTTP API (e.g. http://metacore-app:9000) */
apiUrl: string;
@@ -139,6 +157,27 @@ export class MetaCoreApiWriter {
return Array.isArray(ids) ? ids : [];
}
/**
* GET /api/files/tuples. meta-core's FILE-BACKED roots (records carrying
* filePath, sizeByte and mtimeNano) with sizes, read server-side in batched
* MGETs — the cheap alternative to walking getAllHashIds(), which on a
* shared meta-core is every gateway record too. `summary` omits the rows.
*
* Returns null when meta-core predates the endpoint (404, or a body that is
* not the expected JSON).
*/
async getFileTuples(opts: { summary?: boolean } = {}): Promise<FileTuplesResult | null> {
const path = opts.summary ? '/api/files/tuples?summary=1' : '/api/files/tuples';
const body = await this.fetchJsonAllowing404('GET', path);
if (!body || typeof body !== 'object') return null;
const b = body as { count?: unknown; totalSize?: unknown; files?: unknown };
return {
count: Number(b.count) || 0,
totalSize: Number(b.totalSize) || 0,
files: Array.isArray(b.files) ? (b.files as FileTuple[]) : undefined,
};
}
/**
* GET /health on meta-core. Used as a write-path health probe.
*/
+26 -2
View File
@@ -21,7 +21,7 @@ import {
buildFilePrefix,
buildPropertyKey,
} from './MetadataUtils.js';
import { MetaCoreApiWriter } from './MetaCoreApiWriter.js';
import { MetaCoreApiWriter, type FileTuplesResult } from './MetaCoreApiWriter.js';
// ESM/CJS interop for ioredis
const Redis = (IORedis as any).default ?? IORedis;
@@ -102,7 +102,22 @@ export class RedisKVClient implements IKVClient {
// automatic connect. Any caller that tries to use the Redis
// path (e.g. unmigrated stream code) will fail loudly rather
// than silently misbehaving.
this.redis = new Redis({ host: '127.0.0.1', port: 0, lazyConnect: true, enableOfflineQueue: false });
this.redis = new Redis({
host: '127.0.0.1',
port: 0,
lazyConnect: true,
enableOfflineQueue: false,
// Never reconnect. A stub that retries turns one stray call into
// an endless "Unhandled error event: ECONNREFUSED" loop.
retryStrategy: () => null,
maxRetriesPerRequest: 0,
});
let stubWarned = false;
this.redis.on('error', (err: Error) => {
if (stubWarned) return;
stubWarned = true;
console.warn(`[Redis] HTTP-only stub was used and refused (${err.message}) — a caller still reaches for direct Redis`);
});
this.isConnected = false;
console.log('[Redis] HTTP-only mode — no direct Redis connection');
return;
@@ -474,6 +489,15 @@ export class RedisKVClient implements IKVClient {
return await this.redis.smembers(indexKey);
}
/**
* File-backed roots from meta-core (GET /api/files/tuples). Only the HTTP
* writer can serve this; direct-Redis mode returns null so callers take
* their degraded path instead of sweeping every record.
*/
async getFileTuples(opts: { summary?: boolean } = {}): Promise<FileTuplesResult | null> {
return this.writer ? this.writer.getFileTuples(opts) : null;
}
// ========================================================================
// Pub/Sub Operations
// ========================================================================
@@ -195,6 +195,16 @@ export class UnifiedProcessingStateManager {
performanceMetrics.recordHashProcessing(hashProcessingTime);
}
// File-level metric, one per file that finished cleanly. Nothing recorded
// it before, so the monitor's "Processed" tile stayed at 0. Files announced
// over SSE are never marked discovered, so fall back to the light phase.
if (!error) {
const fileTime = totalProcessingTime
?? (hashState?.lightProcessingStartedAt !== undefined ? now - hashState.lightProcessingStartedAt : hashProcessingTime)
?? 0;
performanceMetrics.recordFileProcessing(filePath, fileTime);
}
const doneState: UnifiedFileState = {
filePath,
state: 'done',
@@ -8,6 +8,7 @@
import type { HashMeta } from '@metazla/meta-interface';
import type { IKVClient } from '../kv/IKVClient.js';
import type { FileTuplesResult } from '../kv/MetaCoreApiWriter.js';
import type { FileAnalyzerInterface } from '../logic/fileProcessor/FileAnalyzerInterface.js';
import type { ContainerTask } from '../container-plugins/types.js';
import type { UnifiedProcessingSnapshot } from '../logic/UnifiedProcessingStateManager.js';
@@ -41,6 +42,21 @@ export function hasPublish(client: IKVClient): client is IKVClientWithPubSub {
return typeof (client as IKVClientWithPubSub).publish === 'function';
}
/**
* IKVClient that can list meta-core's file-backed roots in one call
* (GET /api/files/tuples). Resolves null when meta-core predates the endpoint.
*/
export interface IKVClientWithFileTuples extends IKVClient {
getFileTuples(opts?: { summary?: boolean }): Promise<FileTuplesResult | null>;
}
/**
* Type guard to check if a KVClient can list file-backed roots
*/
export function hasFileTuples(client: IKVClient): client is IKVClientWithFileTuples {
return typeof (client as IKVClientWithFileTuples).getFileTuples === 'function';
}
/**
* IKVClient with Redis Streams capabilities
*
+29 -23
View File
@@ -1,4 +1,4 @@
import { useState, useEffect, useCallback } from 'react';
import { useState, useEffect, useCallback, useRef } from 'react';
import {
ProcessingStatus,
Metrics,
@@ -60,29 +60,35 @@ function Monitor() {
const [failedFiles, setFailedFiles] = useState<FailedFile[]>([]);
const [showPipelineHelp, setShowPipelineHelp] = useState(false);
const fetchData = useCallback(async () => {
try {
const [statusRes, metricsRes, queueRes, statsRes, failedRes] = await Promise.all([
fetch('/api/processing/status'),
fetch('/api/metrics'),
fetch('/api/processing/queue'),
fetch('/api/stats'),
fetch('/api/processing/failed')
]);
// Each endpoint renders as soon as it answers, so one slow or failing call
// cannot blank the page. A tick is skipped while the previous one is still
// out, so a slow backend never collects a pile of overlapping polls.
const pollInFlight = useRef(false);
if (statusRes.ok) setStatus(await statusRes.json());
if (metricsRes.ok) setMetrics(await metricsRes.json());
if (queueRes.ok) {
const queueData = await queueRes.json();
setQueue(queueData.items || []);
const fetchData = useCallback(async () => {
if (pollInFlight.current) return;
pollInFlight.current = true;
const load = async <T,>(url: string, apply: (data: T) => void) => {
const res = await fetch(url);
if (res.ok) apply(await res.json());
};
try {
const results = await Promise.allSettled([
load<ProcessingStatus>('/api/processing/status', setStatus),
load<Metrics>('/api/metrics', setMetrics),
load<{ items?: QueueItem[] }>('/api/processing/queue', (data) => setQueue(data.items || [])),
load<RedisStats>('/api/stats', setRedisStats),
load<{ failedFiles?: FailedFile[] }>('/api/processing/failed', (data) => setFailedFiles(data.failedFiles || [])),
]);
for (const result of results) {
if (result.status === 'rejected') {
console.error('Failed to fetch data:', result.reason);
}
if (statsRes.ok) setRedisStats(await statsRes.json());
if (failedRes.ok) {
const failedData = await failedRes.json();
setFailedFiles(failedData.failedFiles || []);
}
} catch (err) {
console.error('Failed to fetch data:', err);
} finally {
pollInFlight.current = false;
}
}, []);
@@ -410,7 +416,7 @@ function Monitor() {
<h3 className="section-title">File Coverage</h3>
<div className="stats-grid">
<div className="stat-box">
<div className="stat-value">{status?.uniqueHashes ? formatNumber(status.uniqueHashes) : (redisStats ? formatNumber(redisStats.fileCount) : '-')}</div>
<div className="stat-value">{status?.uniqueHashes ? formatNumber(status.uniqueHashes) : (redisStats?.fileCount != null ? formatNumber(redisStats.fileCount) : '-')}</div>
<div className="stat-label">Unique Files</div>
</div>
<div className="stat-box">
@@ -432,7 +438,7 @@ function Monitor() {
<div className="card total-size-card">
<h2>Total Size</h2>
<div className="total-size-value">
{redisStats ? formatBytes(redisStats.totalSize) : '-'}
{redisStats?.totalSize != null ? formatBytes(redisStats.totalSize) : '-'}
</div>
<div className="total-size-label">Watch Folders</div>
</div>
+5 -5
View File
@@ -184,12 +184,12 @@ export interface Metrics {
// NOTE: DuplicateGroup and DuplicateData types have been moved to meta-dup service
/** GET /api/stats: totals over meta-core's file-backed roots. */
export interface RedisStats {
fileCount: number;
keyCount: number;
totalSize: number;
memoryUsage?: string;
memoryUsageBytes?: number;
/** null when meta-core cannot supply file totals (source 'unavailable') */
fileCount: number | null;
totalSize: number | null;
source?: 'meta-core' | 'unavailable';
}
export interface QueueItem {
@@ -1,10 +1,17 @@
/**
* Multicodec / multihash codes for every digest meta-hash can compute.
*
* Source of truth for these numbers, mirrored in:
* Source of truth for these numbers. The same codes appear in the CID utility
* RANK, which is a separate thing this file does not implement:
* - `packages/meta-core/internal/cid/rank.go`
* - the `cid_rank.rs` in each of meta-search and meta-share
* - `/cid-rank-vectors.json` (the golden fixture that pins them all together)
* - `/cid-rank-vectors.json` (the golden fixture pinning those three together)
*
* ⚠ This package used to vendor that fixture at `src/tests/cid-rank-vectors.json`
* and no test ever loaded it, so every vector added cost a commit here for
* nothing. Removed 2026-09-09. If a TypeScript rank implementation is ever
* written, re-add the path to `scripts/check-cid-vectors.sh` *and* a test that
* asserts it — a vendored fixture nobody reads is not a guard.
*
* ⚠ meta-hash encodes `codec == multihash code` (see `ComputeHash.cidFinalize`:
* `CID.createV1(code, create(code, digest))`), whereas the fullhash plugin and
@@ -1,178 +0,0 @@
{
"$schema": "cid-rank-vectors/v1",
"$comment": [
"CANONICAL golden vectors for the CID utility rank. THE single source of truth.",
"The rank decides which of a record's several CIDs is elected canonical - the cid a",
"record is advertised and addressed by. It is reimplemented in Go, 2x Rust and 3x TS;",
"they cannot share code (independent submodules, no cargo path-dep). These vectors are",
"the only thing holding them in sync: every implementation's test suite loads this file.",
"",
"This file is VENDORED into each consuming submodule (submodule CI cannot read above its",
"own tree). scripts/check-cid-vectors.sh diffs every copy against this one.",
"",
"Ranking is by UTILITY for external retrieval, not cryptographic strength.",
"Ties break on the lexicographically smaller cid, so the election is deterministic",
"across peers and does not churn.",
"",
"See METADATA_KEYS.md section 2/14.13 and docs/cid-formats.md."
],
"ladder": {
"40": "dag-pb (IPFS block)",
"30": "sha2-256 / sha3-256 / sha3-384 - real content digests",
"25": "nzb-posting - content-derived (Message-ID set digest), but not itself fetchable",
"20": "BitTorrent infohashes - swarm interop",
"10": "midhash256 - the record address",
"5": "opaque locators (nzb-release, url, card, yt-video, ext-play) - never outrank a real digest",
"0": "weak/unknown digests, unparseable input"
},
"tieBreak": "lexicographically smaller cid wins",
"vectors": [
{
"name": "ipfs-dagpb",
"cid": "bafybeiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 40,
"note": "dag-pb codec 0x70 - chunked IPFS file, broadest network reach"
},
{
"name": "sha2-256-raw-codec",
"cid": "bafkreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 30,
"note": "fullhash/gateway convention: raw codec 0x55, mh names the algo"
},
{
"name": "sha2-256-codec-eq-mh",
"cid": "baejbeibku6ua2l6r5olnggx4pto2sii2s5jeeawgh4cbkhlbmp52gudbua",
"rank": 30,
"note": "REAL meta-hash fixture; meta-hash sets codec == mh"
},
{
"name": "sha3-256",
"cid": "baelbmidner7x3dd7t2kldh6wiq4w6lsye7qrwo6uoqot2t52zkhqk56fnq",
"rank": 30,
"note": "REAL meta-hash fixture"
},
{
"name": "sha3-384",
"cid": "baekrkmeglipkjf33exqexdvijl4twipruay4wsbvt44akce43lq5rgccxgr7v6dwksmj3sxedakevc53g6ga",
"rank": 30,
"note": "REAL meta-hash fixture"
},
{
"name": "nzb-posting",
"cid": "bagbsaeraaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 25,
"note": "codec 0x1003 + REAL sha2-256 multihash (0x12) over the normalised, sorted Message-ID set - NOT a locator, so it sits ABOVE tier 5, but it addresses no fetchable block (bitswap can never satisfy a want for it - the digest is over an id set, not over any block's bytes), so it stays BELOW a real content digest (30): once meta-share materialises the media and re-seeds it, that genuine sha2-256 cid must win the canonical election, not this one. Rank MUST match on codec BEFORE falling through to the mh switch, or this scores 30 (same trap as nzb-release/url, in reverse: those need a codec check because their mh is identity 0x00; this needs one because its mh is REAL sha2-256 and would otherwise tie with tier 30)."
},
{
"name": "btih-v2-codec-eq-mh",
"cid": "bag3sdnzbeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "meta-hash encoder convention"
},
{
"name": "btih-v2-raw-codec",
"cid": "bafk3oijaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "fullhash encoder convention - SAME tier, both conventions must match"
},
{
"name": "btih-v1-file",
"cid": "bagasbajacqaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "gateway per-file BitTorrent v1"
},
{
"name": "btih-v2-file",
"cid": "bagbcbaraeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "gateway per-file BitTorrent v2"
},
{
"name": "midhash256-codec-eq-mh",
"cid": "bagacbabaeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 10,
"note": "the record address; internally meaningful, externally invisible"
},
{
"name": "midhash256-raw-codec",
"cid": "bafkyaibaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 10
},
{
"name": "nzb-release-locator",
"cid": "bagcsaaazcbqxa2jonz5gez3fmvvs42lomzx55ln657fp4aab",
"rank": 5,
"note": "codec 0x1005 + IDENTITY multihash 0x00. The locator code is in the CODEC slot, not the mh slot - rank MUST match on codec. Tier 5 = below midhash: a locator never outranks a real digest, it wins only as a record's sole cid."
},
{
"name": "url-locator",
"cid": "bagdcaaaznb2hi4dthixs6zlymfwxa3dffzxxezzpmexhazdg",
"rank": 5,
"note": "codec 0x1006 + IDENTITY multihash 0x00. Same tier/reasoning as nzb-release."
},
{
"name": "card-locator",
"cid": "bagdsaaanar2g2zdcor3duojvgq3ts",
"rank": 5,
"note": "codec 0x1007 + IDENTITY multihash 0x00, digest = varint(len(source)) || source || id — here (\"tmdb\",\"tv:95479\"). Same codec-slot rule as nzb-release/url: match on the CODEC, never the mh. The limit case of the family — the only CID in MetaMesh with NO bytes behind it anywhere, ever: it addresses a *work* (a series, a film), not a file, so the byte path 404s by design and bitswap never offers it (seeding is gated on sha2-256). Tier 5 is load-bearing for the planned cross-source merge: one card record may carry BOTH a TMDB and a MyAnimeList locator, and if a real card body is ever added (see card-tier-search.md §4.3) its sha2-256 must win the canonical election over both locators without any addressing change. Ranking it 0 instead would still lose to a digest, but would put the implementations back out of agreement — which is the exact failure this fixture exists to prevent."
},
{
"name": "yt-video-locator",
"cid": "bagecaaarav3gszdfn42ein3vgvfumn2tka4a",
"rank": 5,
"note": "codec 0x1008 + IDENTITY multihash 0x00, digest = varint(len(kind)) || kind || id — here (\"video\",\"4D7u5KF7SP8\"). DELEGATED PLAYBACK: the limit case beyond card. A card addresses a work with no bytes anywhere; this addresses a *rendition* whose bytes exist but are permanently someone else's — an external player renders it, nothing is transported, seeded or content-addressed. Framing is byte-identical to card (0x1007) on purpose, so each implementation extends the parser it already wrote. Same codec-slot rule: match on the CODEC, never the mh (identity 0x00 would fall through and score 0). Tier 5 is load-bearing for the on-ramp: a delegated reference and a real archived copy can sit on ONE record, and that sha2-256 must win the canonical election over this locator with no addressing change. NOTE the byte path must REFUSE this (400), not 404 — see docs/cid-formats.md §7.4."
},
{
"name": "ext-play-locator",
"cid": "bagesaabenb2hi4dthixs6zlymfwxa3dffzrgc3temnqw24bomnxw2l3bnrrhk3jppa",
"rank": 5,
"note": "codec 0x1009 + IDENTITY multihash 0x00, digest = the verbatim UTF-8 URL. WIRE-IDENTICAL to url (0x1006) apart from the codec, and that difference is the entire contract: 0x1006 means 'fetch these bytes once and seed them', 0x1009 means 'open this in a browser, there are no bytes for us'. A decoder that matched on SHAPE rather than on the codec slot would see an identity multihash wrapping an http(s) URL, take the 0x1006 path, fetch a watch/Bandcamp page, chunk it, seed the HTML and link that content CID onto the record AS THE FILE — the MSR1 poisoning shape, except permanent and replicated. Same tier/reasoning as the rest of the locator family."
},
{
"name": "sha1",
"cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "md5",
"cid": "bahkqdvibcatpcfl7ine2n625eb5ml5mzkfaa",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "crc32",
"cid": "bagzafmqcathsbnrk",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "bt-pieces-root",
"cid": "bafkyf3qceaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 0,
"note": "torrent-internal merkle root, NOT a file-identity cid"
},
{
"name": "cidv0-rejected",
"cid": "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
"rank": 0,
"note": "TRAP: Go rejected this (rank 0) while Rust's MsCid::from_str ACCEPTED it and ranked it 40 (dag-pb). Strict parse = CIDv1 base32-lower only. This vector pins the fix."
},
{
"name": "token-form-rejected",
"cid": "midhash256:bagacbabaeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 0,
"note": "TRAP: the deprecated <algo>:<cid> token form. Rust used to strip the prefix and rank it 10. It must now be rejected."
},
{
"name": "garbage",
"cid": "not-a-cid",
"rank": 0
},
{
"name": "empty",
"cid": "",
"rank": 0
}
]
}