This commit is contained in:
legop3
2026-09-28 18:15:37 -04:00
parent b59389870e
commit 50c2c1f5db
7 changed files with 79 additions and 23 deletions
+22
View File
@@ -799,3 +799,25 @@ Format 2 snapshots, Latest/Previous, recordings and current training state remai
This supersedes earlier promises to preserve incompatible format 1 model files.
Deletion occurs on the actual server when the updated service starts; no production
files were accessed from this development machine.
### Recording maintenance race and diagnostic error details
Authorized by "fix it and fix that IPI" after a production report showed two
maintenance failures but omitted their error details. Scoped to those two defects;
evaluation/cache changes remain proposals, not part of this repair.
- listRecordings now tolerates ENOENT for directories/files removed between listing
and stat (dock-buffer pruning or atomic metadata rename). Other filesystem
errors still propagate; no broad error suppression or retries were added.
- Diagnostics reportVersion 2 includes actual redacted error messages, rejection
reasons, controller results, event detail, and worker stderr. Runtime catches
retain error name/code/syscall/path/stack, timestamp and operation. Recording
failures retain the same structured details through session finalization.
- A shared redactor removes URLs, credentials, user keys, email addresses and home
usernames while retaining filesystem operation context and stack frames. Error
categories remain supplementary, not replacements for actual failures.
- Node syntax checks passed for affected modules. No actual production recording
stream is available here; the race is fixed by inspection but cannot yet be
confirmed as the cause of the earlier production events. Previously discarded
error details cannot be reconstructed. No new tests or processes were created.
@@ -1,4 +1,5 @@
// Export an explicit allowlist, never the raw socket/session/configuration object.
const { redact } = require('./errors');
const os = require('node:os');
const fs = require('node:fs');
const path = require('node:path');
@@ -6,7 +7,7 @@ const { createHash } = require('node:crypto');
const applicationVersion = require('../../../package.json').version;
const hash = createHash('sha256');
for (const file of ['workers/policy.py', 'workers/dataset.py', 'workers/train.py', 'workers/infer.py',
'models.js', 'driving.js', 'recording.js', 'training.js', 'socketGateway.js', 'diagnostics.js']) {
'errors.js', 'storage.js', 'runtime.js', 'models.js', 'driving.js', 'recording.js', 'training.js', 'socketGateway.js', 'diagnostics.js']) {
hash.update(fs.readFileSync(path.join(__dirname, file)));
}
const implementationSha256 = hash.digest('hex');
@@ -28,7 +29,7 @@ function report(state, controller) {
const metrics = training.diagnostics || {};
const session = state.session;
return {
reportVersion: 1, generatedAtUnixMs: state.updatedAt,
reportVersion: 2, generatedAtUnixMs: state.updatedAt,
software: { applicationVersion, implementationSha256, policyVersion: 2, node: process.version,
dependencies: metrics.dependencies || null },
host: { logicalCpus: os.cpus().length, totalMemoryBytes: os.totalmem(), freeMemoryBytes: os.freemem(), loadAverage: os.loadavg() },
@@ -37,23 +38,23 @@ function report(state, controller) {
freeBytes: state.recording.freeBytes, pendingBytes: state.recording.pendingBytes,
droppedPending: state.recording.droppedPending,
sessions: state.recording.sessions.map((item) => ({ startedAtUnixMs: item.startedAt,
status: item.video, droppedEvents: item.droppedEvents, errorCategory: errorCategory(item.error) })) },
status: item.video, droppedEvents: item.droppedEvents, error: redact(item.error), failureDetails: item.failureDetails || null, stderr: redact(item.stderr), errorCategory: errorCategory(item.error) })) },
training: { status: training.status, steps: training.steps, examples: training.examples,
loss: training.loss, losses: training.losses || null, publication: training.publication || null,
evaluation: training.evaluation || null, dataset: metrics.dataset || training.dataset || null,
resources: { decodeSeconds: metrics.decodeSeconds ?? null, trainingSeconds: metrics.trainingSeconds ?? null,
evaluationSeconds: metrics.evaluationSeconds ?? null, checkpointSeconds: metrics.checkpointSeconds ?? null, cpuSeconds: metrics.cpuSeconds ?? null,
peakRssMiB: metrics.peakRssMiB ?? null },
history: training.history || [], rejectionCategories: (training.rejectionReasons || []).map(errorCategory),
errorCategory: errorCategory(training.error), nextAttemptAtUnixMs: training.nextAttemptAt },
history: training.history || [], rejectionReasons: (training.rejectionReasons || []).map(redact), rejectionCategories: (training.rejectionReasons || []).map(errorCategory),
error: redact(training.error), errorCategory: errorCategory(training.error), nextAttemptAtUnixMs: training.nextAttemptAt },
model: session ? { id: session.modelId, checkpointId: session.checkpointId, threshold: session.threshold, appliedThreshold: session.appliedThreshold } : null,
controller: controller ? { modelId: controller.modelId, sensorPresent: controller.sensorPresent || null,
stopCategory: errorCategory(controller.stopReason), history: (controller.history || []).map((item) => ({
stopReason: redact(controller.stopReason), stopCategory: errorCategory(controller.stopReason), history: (controller.history || []).map((item) => ({
...item, submissions: item.submissions.map((submission) => ({ type: submission.type,
wheels: submission.wheels, result: submission.result === 'issued' ? 'issued' : errorCategory(submission.result) })) })) } : null,
wheels: submission.wheels, result: redact(submission.result) })) })) } : null,
sensorFields: metrics.policy?.sensorFields || null,
timing: { alignment: 'server_receipt', browserVideoLatencyMs: null, wheelCommandExpiryMs: null },
events: state.activity.map(({ at, message }) => ({ atUnixMs: at, message })),
events: state.activity.map(({ at, message, detail, error, stderr }) => ({ atUnixMs: at, message: redact(message), detail: redact(detail), error: error || null, stderr: redact(stderr) })),
unavailable: ['browser_video_latency', 'actual_physical_command_execution', 'live_training_cpu_utilization'],
};
}
@@ -0,0 +1,18 @@
// Keep actionable failures while removing credentials and user identifiers.
function redact(value) {
if (value == null) return null;
return String(value).slice(0, 12000)
.replace(/\b(?:https?|rtsp|rtsps):\/\/[^\s'"<>]+/gi, '[redacted URL]')
.replace(/\b(Bearer\s+)\S+/gi, '$1[redacted]')
.replace(/((?:password|token|secret|authorization|cookie|api[_-]?key)\s*[=:]\s*)[^\s,;]+/gi, '$1[redacted]')
.replace(/\bcu_[a-z0-9]+/gi, '[redacted user]')
.replace(/\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi, '[redacted email]')
.replace(/\/(?:home|Users)\/[^/\s]+/g, '/[user]');
}
function errorDetails(error, operation) {
return { atUnixMs: Date.now(), operation: redact(operation),
name: redact(error?.name), message: redact(error?.message ?? error),
code: redact(error?.code), syscall: redact(error?.syscall),
path: redact(error?.path), stack: redact(error?.stack) };
}
module.exports = { redact, errorDetails };
@@ -1,5 +1,6 @@
// One bounded recording session. Video packets retain receiver wall-clock PTS;
// telemetry uses the same server clock, not an assumed browser playback clock.
const { errorDetails, redact } = require('./errors');
const fs = require('node:fs');
const fsp = require('node:fs/promises');
const path = require('node:path');
@@ -34,6 +35,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
let dockedAt = null;
let pruning = Promise.resolve();
let failure = null;
let failureDetails = null;
let stopping = false;
let stopPromise;
let child;
@@ -43,7 +45,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
let resolveVideo;
const videoClosed = new Promise((resolve) => { resolveVideo = resolve; });
output.on('error', (error) => {
failure = error.message;
failure = error.message; failureDetails = errorDetails(error, 'recording session');
// Stop file growth immediately when telemetry is no longer writable.
child?.kill('SIGTERM');
});
@@ -76,7 +78,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
output.write(line);
} catch (error) {
metadata.droppedEvents += 1;
failure = error.message;
failure = error.message; failureDetails = errorDetails(error, 'recording session');
child?.kill('SIGTERM');
}
return true;
@@ -97,7 +99,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
'-reset_timestamps', '0', path.join(directory, 'video-%09d.mkv'),
], { stdio: ['ignore', 'ignore', 'pipe'] });
child.stderr.on('data', (chunk) => { stderr = (stderr + chunk.toString()).slice(-4096); });
child.on('error', (error) => { failure = error.message; });
child.on('error', (error) => { failure = error.message; failureDetails = errorDetails(error, 'recording session'); });
child.once('close', (code, signal) => {
childClosed = true;
videoResult = { code, signal, stderr };
@@ -131,7 +133,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
if (!metadata.bufferOnly || stopping) return;
if (stat.mtimeMs < Date.now() - 5000) await fsp.unlink(filename);
}
}).catch((error) => { failure = error.message; child.kill('SIGTERM'); });
}).catch((error) => { failure = error.message; failureDetails = errorDetails(error, 'recording session'); child.kill('SIGTERM'); });
}
async function stop(reason) {
@@ -158,9 +160,10 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
metadata.endedAt = Date.now();
metadata.reason = reason;
metadata.error = failure;
metadata.failureDetails = failureDetails;
metadata.video.result = videoResult;
await saveMetadata(directory, metadata);
if (failure) logger.warn('Recording session incomplete', { roverId, id, error: failure, stderr });
if (failure) logger.warn('Recording session incomplete', { roverId, id, error: failure, failure: failureDetails, stderr });
})();
return stopPromise;
}
@@ -173,7 +176,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
get failed() { return Boolean(failure); },
get stopping() { return stopping; },
getStatus: () => ({ id, roverId, startedAt: metadata.startedAt, droppedEvents: metadata.droppedEvents,
video: childClosed ? 'stopped' : metadata.bufferOnly ? 'Dock lead-in buffer' : 'Recording human driving', error: failure, stopping }),
video: childClosed ? 'stopped' : metadata.bufferOnly ? 'Dock lead-in buffer' : 'Recording human driving', error: redact(failure), failureDetails, stderr: redact(stderr), stopping }),
// Process exit is synchronous; normal disable uses stop() and awaits close.
kill: () => { if (!childClosed) child.kill('SIGKILL'); },
};
@@ -1,4 +1,5 @@
// Enabled-only recording, training, controller, and card lifecycle.
const { redact, errorDetails } = require('./errors');
const fs = require('node:fs/promises');
const path = require('node:path');
const { resolveDataPath } = require('../../helpers/dataPaths');
@@ -22,7 +23,9 @@ function createRuntime({ config, logger }) {
const recentActivity = [];
function activity(message, details = {}) {
recentActivity.push({ at: Date.now(), message, roverId: details.roverId || null,
detail: details.reason || details.error || details.model || null });
detail: redact(details.reason || details.error || details.model),
error: details.failure || (details.error ? errorDetails(details.error, message) : null),
stderr: redact(details.stderr) });
if (recentActivity.length > 40) recentActivity.shift();
}
const baseLogger = logger;
@@ -77,7 +80,7 @@ function createRuntime({ config, logger }) {
// The serialized maintenance pass still owns closing/removing the session.
for (const [id, session] of sessions) {
if (!next.has(id)) session.stop('no-human-controller').catch((error) => {
logger.warn('Unable to finalize unattended recording', { roverId: id, error: error.message });
logger.warn('Unable to finalize unattended recording', { roverId: id, error: error.message, failure: errorDetails(error, 'rover learning runtime') });
});
}
schedule();
@@ -185,7 +188,7 @@ function createRuntime({ config, logger }) {
await trainer.tick(new Set([...sessions.values()].map((session) => session.id)));
} catch (error) {
// Training availability must not tear down otherwise healthy capture.
logger.error('Unable to schedule training', { error: error.message });
logger.error('Unable to schedule training', { error: error.message, failure: errorDetails(error, 'rover learning runtime') });
}
}
}
@@ -198,7 +201,7 @@ function createRuntime({ config, logger }) {
// The next tick is scheduled after completion: disk slowness cannot build
// an unbounded backlog of retention tasks or overlapping recorder starts.
maintenance = maintain().catch(async (error) => {
logger.error('Recording maintenance failed; closing recorders', { error: error.message });
logger.error('Recording maintenance failed; closing recorders', { error: error.message, failure: errorDetails(error, 'recording maintenance') });
await Promise.allSettled([...sessions.keys()].map((id) => closeSession(id, 'storage-error')));
}).finally(() => {
busy = false;
@@ -216,7 +219,7 @@ function createRuntime({ config, logger }) {
// Other server services can exit the process before asynchronous cleanup
// finishes. Signal children immediately, and keep an exit fallback too.
for (const session of sessions.values()) session.stop('server-shutdown').catch(() => undefined);
stop().catch((error) => logger.warn('Recording shutdown failed', { error: error.message }));
stop().catch((error) => logger.warn('Recording shutdown failed', { error: error.message, failure: errorDetails(error, 'rover learning runtime') }));
}
async function start() {
@@ -10,9 +10,17 @@ async function listRecordings(root) {
if (!entry.isDirectory() || !/^\d+-[0-9a-f-]{36}$/.test(entry.name)) continue;
const directory = path.join(root, entry.name);
let bytes = 0;
for (const file of await fs.readdir(directory, { withFileTypes: true })) {
let files;
try { files = await fs.readdir(directory, { withFileTypes: true }); }
catch (error) { if (error.code === 'ENOENT') continue; throw error; }
for (const file of files) {
if (!file.isFile()) continue;
bytes += (await fs.stat(path.join(directory, file.name))).size;
try { bytes += (await fs.stat(path.join(directory, file.name))).size; }
catch (error) {
// Dock-buffer pruning and atomic metadata replacement can remove a
// listed file before stat. Absence is normal; other I/O failures are not.
if (error.code !== 'ENOENT') throw error;
}
}
recordings.push({ id: entry.name, directory, bytes, startedAt: Number(entry.name.split('-')[0]) });
}
@@ -1,5 +1,6 @@
// One owned finite worker at a time. Leases are acquired in the runtime's
// serialized maintenance pass, before retention can next inspect recordings.
const { errorDetails } = require('./errors');
const fs = require('node:fs/promises');
const path = require('node:path');
const { spawn } = require('node:child_process');
@@ -40,7 +41,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
} catch (error) {
startupError = true;
state = { ...state, status: 'error', error: error.message };
logger.error('Training initialization failed; recording remains available', { error: error.message });
logger.error('Training initialization failed; recording remains available', { error: error.message, failure: errorDetails(error, 'training initialization') });
}
}
@@ -148,7 +149,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
}
} catch (error) {
state = { ...state, status: 'error', error: error.message };
logger.error('Training result persistence failed', { error: error.message });
logger.error('Training result persistence failed', { error: error.message, failure: errorDetails(error, 'training result persistence') });
} finally {
history.push({ at: Date.now(), status: state.status, steps: state.steps, loss: state.loss,
losses: state.losses || null, evaluation: state.evaluation || null,