fix(firmware,ssh,logs): five reported bugs (v0.24.1-beta)
Closes #136. Also fixes #133, #137, #138, #139 (GitHub closes only the first). None of these came from multi-site; two came from me. #136 — waves failed on the second device. A regression from v0.23.31, the firmware rewrite that replaced fire-and-forget `install` with `download` plus polling. `/system/package/update/download` blocks for a minute or more, and the client's read timeout is 30 seconds. The timeout path was the real damage: it nulled the pending read but left the socket open, so when the reply finally arrived processBuffer() queued it and the *next* command took it as its own answer. Every later call on that connection returned the previous one's data, which is why the status polls could never see "Downloaded". - a timed-out read now abandons the connection instead of silently desynchronising it, so the failure is loud rather than wrong - long-running commands can state their own budget; check-for-updates, download and /log/print now do - a download whose command times out is re-checked on a fresh connection rather than assumed failed, because the image may well have landed #133 — the SSH terminal could not use a deployed key. It read only ssh_password_encrypted and refused to connect without one, so retiring the password after deploying a key -- the workflow the feature exists for -- broke the console. It now uses the same resolver as every other SSH consumer. preferredAuth(), written to make exactly this decision, had never run outside its own tests, and disagreed with the resolver that did: it accepted a 'deployed' key where the live path required 'verified'. Unified on the stricter rule and wired in. #139 — the rollout was written before start() was attempted, so a rejected start left the rows behind: "already running", press again, another rollout banked each time. Devices already in an unfinished rollout are now refused by name, before anything is persisted. #138 — cancel worked but the API's explanation that the in-flight device is never interrupted was discarded, so the button looked dead. It is shown. #137 — log overfetching. /log/print returns the whole ring buffer, and the count was incremented per attempted insert rather than per accepted row, so a poll that stored nothing still reported "Collected 1000 new log entries" after a thousand round trips. Inserts are batched and counted by RETURNING, and the id watermark is the maximum over recent rows rather than the last one inserted, which could move backwards and make the whole buffer look new. One self-inflicted wound worth recording. The first cut of the desync guard discarded sentences whenever `connected` was false -- but `connected` is only set *after* login succeeds, so it threw away every login reply and took the whole fleet offline until it was caught. "Not yet connected" and "abandoned" are different states and now have different flags. Every existing test pre-set connected = true, so none of them could see it; the test added for it fails against the bad version. Verified on hardware: devices back online, check-update 5.2s (was a 30s failure), full sync completes, and log collection now reports 298 / 22 / 12 / 12 across successive polls -- real decreasing counts rather than a constant 1000 -- with no failures. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2GT-Rich
co-authored by
Claude Opus 5
Claude Opus 5
parent
7b994ad935
commit
63b5677522
@@ -19,6 +19,18 @@ import {
|
||||
import { bytesSince, periodKey, shouldSend } from '../../utils/dataCap';
|
||||
import { parseDeviceLogTime } from '../../utils/deviceTime';
|
||||
import { BackupService } from '../BackupService';
|
||||
|
||||
/** RouterOS update commands reach out to MikroTik's servers and are slow by nature. */
|
||||
const UPDATE_CHECK_TIMEOUT_MS = 90_000;
|
||||
const UPDATE_DOWNLOAD_TIMEOUT_MS = 10 * 60_000;
|
||||
/**
|
||||
* `/log/print` returns the device's entire ring buffer -- a thousand sentences
|
||||
* is normal -- and measured against real hardware it routinely outruns the 30s
|
||||
* default. Timing it out cost the device's whole log *and*, before the client
|
||||
* learned to abandon a timed-out connection, desynchronised the session (#137).
|
||||
*/
|
||||
const LOG_READ_TIMEOUT_MS = 120_000;
|
||||
import { selectNewLogLines, highestStoredId, type RawLogLine } from '../../utils/logDedup';
|
||||
import { alertService } from '../AlertService';
|
||||
|
||||
/** DB column limits for topology_links (see migrate.ts); reject oversize rows instead of silent truncation. */
|
||||
@@ -800,22 +812,21 @@ export class DeviceCollector {
|
||||
|
||||
async collectEvents(): Promise<void> {
|
||||
try {
|
||||
const logs = await this.client.execute('/log/print');
|
||||
const logs = await this.client.execute(
|
||||
'/log/print', {}, [], { timeoutMs: LOG_READ_TIMEOUT_MS }
|
||||
);
|
||||
if (!logs.length) return;
|
||||
|
||||
// RouterOS log .id values are hex strings like "*1A2F". Parse to int for comparison.
|
||||
const parseRosId = (id: string): number => {
|
||||
const hex = (id || '').replace(/^\*/, '');
|
||||
return hex ? parseInt(hex, 16) : 0;
|
||||
};
|
||||
|
||||
// Find the highest RouterOS log ID we've already stored for this device.
|
||||
const lastRow = await queryOne<{ log_id: string }>(
|
||||
// The highest id we hold, taken across recent rows rather than from the
|
||||
// single most recently inserted one. Those are usually the same, but when
|
||||
// they differ the watermark slips backwards and the entire buffer looks
|
||||
// new on every poll (#137).
|
||||
const storedIds = await query<{ log_id: string }>(
|
||||
`SELECT log_id FROM events WHERE device_id = $1 AND log_id IS NOT NULL AND log_id != ''
|
||||
ORDER BY id DESC LIMIT 1`,
|
||||
ORDER BY id DESC LIMIT 2000`,
|
||||
[this.device.id]
|
||||
);
|
||||
const lastIdNum = parseRosId(lastRow?.log_id || '');
|
||||
const lastIdNum = highestStoredId(storedIds.map((r) => r.log_id));
|
||||
|
||||
// Always get the latest stored event time — needed for timestamp fallback.
|
||||
const latestStored = await queryOne<{ event_time: Date }>(
|
||||
@@ -824,45 +835,55 @@ export class DeviceCollector {
|
||||
);
|
||||
const latestTime = latestStored?.event_time ? new Date(latestStored.event_time) : new Date(0);
|
||||
|
||||
// Detect log buffer overflow or device reboot: if all current IDs are below
|
||||
// our stored lastIdNum, RouterOS has cleared/reset its log buffer.
|
||||
// In that case, fall back to timestamp-based deduplication to avoid missing events.
|
||||
const currentIds = logs.map(l => parseRosId(l['.id'] || '')).filter(id => id > 0);
|
||||
const maxCurrentId = currentIds.length > 0 ? Math.max(...currentIds) : 0;
|
||||
const logReset = lastRow && maxCurrentId > 0 && maxCurrentId < lastIdNum;
|
||||
|
||||
// Collect the candidates first, then insert them in batches.
|
||||
//
|
||||
// This used to issue one INSERT per line and count every attempt as new,
|
||||
// so a poll that re-read the whole ring buffer and inserted nothing still
|
||||
// reported "Collected 1000 new log entries" — a thousand round trips a
|
||||
// minute per device, described as work that had not happened (#137).
|
||||
// Counting what the database actually accepted makes the log line true
|
||||
// and the duplicate case cheap.
|
||||
const fresh = selectNewLogLines(logs as RawLogLine[], {
|
||||
lastStoredId: lastIdNum,
|
||||
latestStoredTime: latestTime,
|
||||
parseTime: (raw) => this.parseLogTime(raw),
|
||||
});
|
||||
|
||||
const pending: unknown[][] = [];
|
||||
for (const log of fresh) {
|
||||
const logId = (log['.id'] || '') as string;
|
||||
const time = this.parseLogTime(log['time'] || '');
|
||||
const severity = this.mapLogSeverity((log['topics'] as string) || '');
|
||||
pending.push([
|
||||
this.device.id,
|
||||
time.toISOString(),
|
||||
severity,
|
||||
(log['topics'] as string) || null,
|
||||
(log['message'] as string) || '',
|
||||
JSON.stringify(log),
|
||||
logId || null,
|
||||
]);
|
||||
}
|
||||
|
||||
let newCount = 0;
|
||||
for (const log of logs) {
|
||||
const logId = (log['.id'] || '') as string;
|
||||
const logIdNum = parseRosId(logId);
|
||||
|
||||
if (logId && !logReset) {
|
||||
// Primary: skip entries already stored by RouterOS ID
|
||||
if (logIdNum <= lastIdNum) continue;
|
||||
} else {
|
||||
// Fallback: no .id field, or log buffer has been reset — use timestamp deduplication
|
||||
const time = this.parseLogTime(log['time'] || '');
|
||||
if (time <= latestTime) continue;
|
||||
}
|
||||
|
||||
const time = this.parseLogTime(log['time'] || '');
|
||||
const severity = this.mapLogSeverity(log['topics'] || '');
|
||||
|
||||
await query(
|
||||
const COLS = 7;
|
||||
const BATCH = 200;
|
||||
for (let i = 0; i < pending.length; i += BATCH) {
|
||||
const chunk = pending.slice(i, i + BATCH);
|
||||
const values = chunk
|
||||
.map((_, r) => `(${Array.from({ length: COLS }, (_, c) => `$${r * COLS + c + 1}`).join(',')})`)
|
||||
.join(',');
|
||||
// RETURNING reports only the rows actually written; ON CONFLICT rows are
|
||||
// silently dropped, which is exactly the count we want.
|
||||
const inserted = await query<{ id: number }>(
|
||||
`INSERT INTO events (device_id, event_time, severity, topic, message, raw_json, log_id)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7)
|
||||
ON CONFLICT (device_id, log_id) DO NOTHING`,
|
||||
[
|
||||
this.device.id,
|
||||
time.toISOString(),
|
||||
severity,
|
||||
log['topics'] || null,
|
||||
log['message'] || '',
|
||||
JSON.stringify(log),
|
||||
logId || null,
|
||||
]
|
||||
VALUES ${values}
|
||||
ON CONFLICT (device_id, log_id) DO NOTHING
|
||||
RETURNING id`,
|
||||
chunk.flat()
|
||||
);
|
||||
newCount++;
|
||||
newCount += inserted.length;
|
||||
}
|
||||
|
||||
if (newCount > 0) {
|
||||
@@ -3757,12 +3778,16 @@ export class DeviceCollector {
|
||||
// ─── Firmware updates ─────────────────────────────────────────────────────
|
||||
|
||||
async checkForUpdates(): Promise<Record<string, string>> {
|
||||
await this.client.execute('/system/package/update/check-for-updates').catch(() => {});
|
||||
// Give the device a moment to complete the check
|
||||
// Reaches out to MikroTik's servers, so it can outlast the ordinary read
|
||||
// timeout on a slow or filtered link. The failure was previously swallowed
|
||||
// (.catch(() => {})), which hid a timed-out check and left the caller
|
||||
// reading whatever /print happened to return (#136).
|
||||
await this.client.execute(
|
||||
'/system/package/update/check-for-updates', {}, [], { timeoutMs: UPDATE_CHECK_TIMEOUT_MS }
|
||||
);
|
||||
// Give the device a moment to settle the result before reading it back.
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 3000));
|
||||
const result = await this.client
|
||||
.execute('/system/package/update/print')
|
||||
.catch(() => [] as Record<string, string>[]);
|
||||
const result = await this.client.execute('/system/package/update/print');
|
||||
return (result[0] as Record<string, string>) || {};
|
||||
}
|
||||
|
||||
@@ -3776,8 +3801,11 @@ export class DeviceCollector {
|
||||
* be observed — and on at least one CCR it neither downloaded nor rebooted
|
||||
* while reporting nothing at all (#firmware).
|
||||
*/
|
||||
async downloadUpdate(): Promise<void> {
|
||||
await this.client.execute('/system/package/update/download');
|
||||
async downloadUpdate(timeoutMs = UPDATE_DOWNLOAD_TIMEOUT_MS): Promise<void> {
|
||||
// Blocks until the image has landed -- a minute or more is normal, and the
|
||||
// 30-second default read timeout was cutting it off and corrupting the
|
||||
// connection (#136).
|
||||
await this.client.execute('/system/package/update/download', {}, [], { timeoutMs });
|
||||
}
|
||||
|
||||
/** The `status` line from /system/package/update, e.g. "Downloaded, please reboot…". */
|
||||
|
||||
@@ -40,6 +40,8 @@ export class RouterOSClient extends EventEmitter {
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
} | null = null;
|
||||
private sentenceQueue: RouterOSSentence[] = [];
|
||||
/** Set when a reply stream can no longer be trusted; distinct from `connected`. */
|
||||
private poisoned = false;
|
||||
|
||||
private tagCounter = 0;
|
||||
|
||||
@@ -62,6 +64,10 @@ export class RouterOSClient extends EventEmitter {
|
||||
|
||||
async connect(): Promise<void> {
|
||||
if (this.connected) return;
|
||||
// A fresh connection starts trusted, so a client that was abandoned earlier
|
||||
// can be reconnected and used again.
|
||||
this.poisoned = false;
|
||||
this.sentenceQueue = [];
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
@@ -124,10 +130,18 @@ export class RouterOSClient extends EventEmitter {
|
||||
}
|
||||
|
||||
// Enqueue a command so concurrent calls don't interleave
|
||||
/**
|
||||
* @param opts.timeoutMs How long to wait for each reply sentence. The 30s
|
||||
* default suits ordinary reads, but some RouterOS commands legitimately run
|
||||
* far longer -- `/system/package/update/download` takes a minute or more on
|
||||
* an ordinary link -- and timing those out corrupted the connection (#136).
|
||||
* Callers issuing a long-running command must say so.
|
||||
*/
|
||||
async execute(
|
||||
command: string,
|
||||
params: Record<string, string> = {},
|
||||
queries: string[] = []
|
||||
queries: string[] = [],
|
||||
opts: { timeoutMs?: number } = {}
|
||||
): Promise<Record<string, string>[]> {
|
||||
let result!: Record<string, string>[];
|
||||
let error!: Error;
|
||||
@@ -135,7 +149,7 @@ export class RouterOSClient extends EventEmitter {
|
||||
await new Promise<void>((resolve) => {
|
||||
this.operationChain = this.operationChain.then(async () => {
|
||||
try {
|
||||
result = await this._executeRaw(command, params, queries);
|
||||
result = await this._executeRaw(command, params, queries, opts.timeoutMs);
|
||||
} catch (err) {
|
||||
error = err as Error;
|
||||
}
|
||||
@@ -150,7 +164,8 @@ export class RouterOSClient extends EventEmitter {
|
||||
private async _executeRaw(
|
||||
command: string,
|
||||
params: Record<string, string>,
|
||||
queries: string[]
|
||||
queries: string[],
|
||||
timeoutMs?: number
|
||||
): Promise<Record<string, string>[]> {
|
||||
if (!this.socket || !this.connected) {
|
||||
throw new RouterOSError('Not connected');
|
||||
@@ -168,7 +183,7 @@ export class RouterOSClient extends EventEmitter {
|
||||
|
||||
const results: Record<string, string>[] = [];
|
||||
while (true) {
|
||||
const sentence = await this.readNextSentence();
|
||||
const sentence = await this.readNextSentence(timeoutMs);
|
||||
if (sentence.type === '!done') break;
|
||||
if (sentence.type === '!re') {
|
||||
results.push(sentence.words);
|
||||
@@ -341,6 +356,17 @@ export class RouterOSClient extends EventEmitter {
|
||||
const sentence = this.tryParseSentence();
|
||||
if (!sentence) break;
|
||||
|
||||
// Once a connection is abandoned, anything still arriving belongs to the
|
||||
// command that gave up. Queueing it would hand it to the next caller as
|
||||
// that caller's own reply, which is the desynchronisation poison() exists
|
||||
// to prevent -- destroying the socket usually stops this, but data already
|
||||
// buffered can still surface.
|
||||
//
|
||||
// This must test `poisoned`, not `connected`: `connected` is only set once
|
||||
// login has *finished*, so testing it here discarded the login exchange
|
||||
// itself and every connection timed out.
|
||||
if (this.poisoned) continue;
|
||||
|
||||
if (this.pendingRead) {
|
||||
clearTimeout(this.pendingRead.timeout);
|
||||
const { resolve } = this.pendingRead;
|
||||
@@ -425,6 +451,19 @@ export class RouterOSClient extends EventEmitter {
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.pendingRead = null;
|
||||
// A timed-out read leaves a reply in flight. Previously the socket was
|
||||
// left open, so when that reply finally arrived processBuffer() found no
|
||||
// pending read and queued it -- and the *next* command took it as its
|
||||
// own answer. Every later command on that connection then returned the
|
||||
// previous one's data.
|
||||
//
|
||||
// That is how a slow `/system/package/update/download` turned into
|
||||
// "did not finish downloading within 10 minutes" (#136): the status
|
||||
// polls afterwards were reading shifted replies and could never see
|
||||
// "Downloaded". A desynchronised connection can also make an unrelated
|
||||
// command look like it succeeded, so the only safe response is to stop
|
||||
// using it.
|
||||
this.poison('Read timeout waiting for API response');
|
||||
reject(new RouterOSError('Read timeout waiting for API response'));
|
||||
}, timeoutMs);
|
||||
|
||||
@@ -436,6 +475,22 @@ export class RouterOSClient extends EventEmitter {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Abandon a connection whose reply stream can no longer be trusted.
|
||||
*
|
||||
* Anything already queued belongs to a command that gave up, so it is
|
||||
* discarded rather than handed to the next caller.
|
||||
*/
|
||||
private poison(reason: string): void {
|
||||
this.poisoned = true;
|
||||
this.connected = false;
|
||||
this.sentenceQueue = [];
|
||||
this.buffer = Buffer.alloc(0);
|
||||
try { this.socket?.destroy(); } catch { /* already gone */ }
|
||||
this.socket = null;
|
||||
this.emit('poisoned', reason);
|
||||
}
|
||||
|
||||
// Execute a streaming/generator command (one that never sends !done on its own,
|
||||
// e.g. /tool/ip-scan). Collects !re rows until maxDurationMs elapses, then
|
||||
// sends /cancel to terminate the command and drains the cleanup response.
|
||||
|
||||
@@ -77,3 +77,91 @@ describe('sentence parsing', () => {
|
||||
expect(REPEATED_KEY).toBe(REPEATED_ATTRIBUTES_KEY);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Read-timeout handling (#136).
|
||||
*
|
||||
* A timed-out read leaves a reply in flight. The connection used to stay open,
|
||||
* so when that reply eventually arrived it was queued and handed to the *next*
|
||||
* command as its own answer — every subsequent call on that connection returned
|
||||
* the previous one's data.
|
||||
*
|
||||
* That is what turned a slow `/system/package/update/download` into "did not
|
||||
* finish downloading within 10 minutes": the status polls afterwards were
|
||||
* reading shifted replies and could never see "Downloaded". The same
|
||||
* desynchronisation could make any unrelated command appear to succeed, so
|
||||
* these tests matter well beyond firmware.
|
||||
*/
|
||||
describe('read timeout', () => {
|
||||
type Internals = {
|
||||
buffer: Buffer;
|
||||
connected: boolean;
|
||||
socket: unknown;
|
||||
sentenceQueue: RouterOSSentence[];
|
||||
pendingRead: unknown;
|
||||
readNextSentence(timeoutMs?: number): Promise<RouterOSSentence>;
|
||||
processBuffer(): void;
|
||||
};
|
||||
|
||||
const makeClient = () => {
|
||||
const client = new RouterOSClient('127.0.0.1', 8728, 'x', 'x');
|
||||
const internals = client as unknown as Internals;
|
||||
internals.connected = true;
|
||||
internals.socket = { destroy() {} };
|
||||
return { client, internals };
|
||||
};
|
||||
|
||||
// The regression these were written without: the discard guard first tested
|
||||
// `connected`, which is only set *after* login succeeds. Every login reply was
|
||||
// therefore thrown away and every connection timed out — the whole fleet went
|
||||
// offline. The tests above all pre-set connected = true, so none of them saw
|
||||
// it. "Not yet connected" and "abandoned" are different states.
|
||||
it('delivers replies that arrive before login has completed', () => {
|
||||
const { internals } = makeClient();
|
||||
internals.connected = false; // mid-login, as connect() leaves it
|
||||
internals.buffer = sentence('ret=abc123');
|
||||
internals.processBuffer();
|
||||
expect(internals.sentenceQueue).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('rejects when no reply arrives in time', async () => {
|
||||
const { internals } = makeClient();
|
||||
await expect(internals.readNextSentence(20)).rejects.toThrow(/Read timeout/);
|
||||
});
|
||||
|
||||
it('does not hand a late reply to the next command', async () => {
|
||||
const { internals } = makeClient();
|
||||
|
||||
await expect(internals.readNextSentence(20)).rejects.toThrow(/Read timeout/);
|
||||
|
||||
// The reply the timed-out command was waiting for now turns up.
|
||||
internals.buffer = sentence('status=Downloaded, please reboot');
|
||||
internals.processBuffer();
|
||||
|
||||
// It must not be sitting in the queue waiting to be mistaken for the next
|
||||
// command's answer.
|
||||
expect(internals.sentenceQueue).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('marks the connection unusable so it cannot be reused blind', async () => {
|
||||
const { client, internals } = makeClient();
|
||||
await expect(internals.readNextSentence(20)).rejects.toThrow(/Read timeout/);
|
||||
|
||||
expect(internals.connected).toBe(false);
|
||||
// A caller that carries on regardless gets a clean error rather than
|
||||
// another command's data.
|
||||
await expect(client.execute('/system/resource/print')).rejects.toThrow(/Not connected/);
|
||||
});
|
||||
|
||||
// The guard above must not break ordinary pipelining: a healthy connection
|
||||
// legitimately queues replies that arrive faster than they are read.
|
||||
it('still serves queued replies while the connection is healthy', async () => {
|
||||
const { internals } = makeClient();
|
||||
internals.buffer = sentence('status=ok');
|
||||
internals.processBuffer();
|
||||
expect(internals.sentenceQueue).toHaveLength(1);
|
||||
await expect(internals.readNextSentence(20)).resolves.toMatchObject({
|
||||
words: { status: 'ok' },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user