fix(notifications): prevent self-container stack routing (#1242)

* fix(notifications): prevent self-container stack routing

* fix(stack-files): stabilize download metrics in CI
This commit is contained in:
Anso
2026-05-28 12:45:10 -04:00
committed by GitHub
parent e57dbb8da7
commit 265fece988
9 changed files with 704 additions and 53 deletions
@@ -25,6 +25,7 @@ const {
mockInspect,
mockGetContainer,
mockGetDocker,
mockIsOwnContainer,
} = vi.hoisted(() => ({
mockDispatchAlert: vi.fn().mockResolvedValue(undefined),
mockBroadcastEvent: vi.fn(),
@@ -34,6 +35,7 @@ const {
mockInspect: vi.fn().mockResolvedValue({}),
mockGetContainer: vi.fn(),
mockGetDocker: vi.fn(),
mockIsOwnContainer: vi.fn().mockReturnValue(false),
}));
vi.mock('../services/NotificationService', () => ({
@@ -57,6 +59,14 @@ vi.mock('../services/NodeRegistry', () => ({
},
}));
vi.mock('../services/SelfIdentityService', () => ({
default: {
getInstance: () => ({
isOwnContainer: mockIsOwnContainer,
}),
},
}));
// ── Fake Docker stream helper ──────────────────────────────────────────
interface FakeStream extends EventEmitter {
@@ -88,6 +98,8 @@ beforeEach(() => {
vi.clearAllMocks();
vi.useFakeTimers();
mockGetGlobalSettings.mockReturnValue({ global_crash: '1' });
mockIsOwnContainer.mockReset();
mockIsOwnContainer.mockReturnValue(false);
stream = makeStream();
mockGetEvents.mockImplementation(async () => stream);
@@ -130,6 +142,61 @@ describe('DockerEventService - die classification', () => {
);
});
it('keeps stack and container routing for non-self compose crashes', async () => {
service = new DockerEventService(1, 'local');
await service.start();
stream.push({
Type: 'container',
Action: 'die',
Actor: {
ID: 'app-id',
Attributes: {
exitCode: '1',
name: 'web',
'com.docker.compose.project': 'web-stack',
},
},
});
await vi.advanceTimersByTimeAsync(600);
expect(mockDispatchAlert).toHaveBeenCalledWith(
'error',
'monitor_alert',
expect.stringContaining('Container Crash Detected'),
expect.objectContaining({ stackName: 'web-stack', containerName: 'web' }),
);
});
it('routes self-container crashes as system-only notifications', async () => {
mockIsOwnContainer.mockImplementation((idOrName: string) =>
idOrName === 'self-id' || idOrName === 'sencho',
);
service = new DockerEventService(1, 'local');
await service.start();
stream.push({
Type: 'container',
Action: 'die',
Actor: {
ID: 'self-id',
Attributes: {
exitCode: '1',
name: 'sencho',
'com.docker.compose.project': 'sencho',
},
},
});
await vi.advanceTimersByTimeAsync(600);
expect(mockDispatchAlert).toHaveBeenCalledWith(
'error',
'monitor_alert',
expect.stringContaining('Container Crash Detected'),
{ actor: 'system:docker-events' },
);
});
it('does not emit when die follows a recent kill (intentional)', async () => {
service = new DockerEventService(1, 'local');
await service.start();
@@ -231,6 +298,33 @@ describe('DockerEventService - die classification', () => {
);
});
it('routes self-container unhealthy alerts as system-only notifications', async () => {
mockIsOwnContainer.mockImplementation((idOrName: string) =>
idOrName === 'self-id' || idOrName === 'sencho',
);
service = new DockerEventService(1, 'local');
await service.start();
stream.push({
Type: 'container',
Action: 'health_status: unhealthy',
Actor: {
ID: 'self-id',
Attributes: {
name: 'sencho',
'com.docker.compose.project': 'sencho',
},
},
});
expect(mockDispatchAlert).toHaveBeenCalledWith(
'error',
'monitor_alert',
expect.stringContaining('Healthcheck failed'),
{ actor: 'system:docker-events' },
);
});
it('does not emit when global_crash is disabled', async () => {
mockGetGlobalSettings.mockReturnValue({ global_crash: '0' });
service = new DockerEventService(1, 'local');
@@ -616,6 +710,30 @@ describe('DockerEventService - state-invalidate broadcasts', () => {
}));
});
it('does not broadcast stack state-invalidate for Sencho self-container events', async () => {
mockIsOwnContainer.mockImplementation((idOrName: string) =>
idOrName === 'self-id' || idOrName === 'sencho',
);
service = new DockerEventService(7, 'node-7');
await service.start();
stream.push({
Type: 'container',
Action: 'start',
Actor: {
ID: 'self-id',
Attributes: {
name: 'sencho',
'com.docker.compose.project': 'sencho',
},
},
time: 1,
});
await vi.advanceTimersByTimeAsync(1);
expect(mockBroadcastEvent).not.toHaveBeenCalled();
});
it('broadcasts state-invalidate on health_status:unhealthy', async () => {
service = new DockerEventService(1, 'local');
await service.start();
@@ -65,6 +65,7 @@ describe('SelfIdentityService.initialize', () => {
Id: FULL_CONTAINER_ID,
Name: '/sencho',
Image: 'sha256:' + FULL_IMAGE_ID_HEX,
Config: { Labels: { 'com.docker.compose.project': 'sencho' } },
NetworkSettings: {
Networks: {
sencho_mesh: { NetworkID: FULL_NETWORK_ID },
@@ -82,11 +83,38 @@ describe('SelfIdentityService.initialize', () => {
const id = svc.getIdentity();
expect(id.containerId).toBe(FULL_CONTAINER_ID);
expect(id.containerName).toBe('sencho');
expect(id.composeProjectName).toBe('sencho');
expect(id.imageId).toBe(FULL_IMAGE_ID_HEX);
expect(id.networkNames).toEqual(['sencho_mesh']);
expect(id.volumeNames).toEqual(['sencho_data']);
});
it('shares an in-flight initialization across concurrent callers', async () => {
process.env.HOSTNAME = 'sencho-1';
let resolveInspect: (value: unknown) => void = () => {};
mockContainer.inspect.mockReturnValue(new Promise(resolve => {
resolveInspect = resolve;
}));
const svc = SelfIdentityService.getInstance();
const first = svc.initialize();
const second = svc.initialize();
expect(mockContainer.inspect).toHaveBeenCalledTimes(1);
resolveInspect({
Id: FULL_CONTAINER_ID,
Name: '/sencho',
Image: 'sha256:' + FULL_IMAGE_ID_HEX,
Config: { Labels: { 'com.docker.compose.project': 'sencho' } },
NetworkSettings: { Networks: { sencho_mesh: { NetworkID: FULL_NETWORK_ID } } },
Mounts: [{ Type: 'volume', Name: 'sencho_data' }],
});
await Promise.all([first, second]);
expect(svc.getIdentity().containerId).toBe(FULL_CONTAINER_ID);
expect(svc.getIdentity().composeProjectName).toBe('sencho');
});
it('stays empty when HOSTNAME is unset (dev mode)', async () => {
delete process.env.HOSTNAME;
const svc = SelfIdentityService.getInstance();
+73 -2
View File
@@ -3,7 +3,7 @@ import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { sanitizeNotificationMessage } from '../utils/notificationMessage';
let tmpDir: string;
let db: any;
let db: import('../services/DatabaseService').DatabaseService;
beforeAll(async () => {
tmpDir = await setupTestDb();
@@ -128,10 +128,13 @@ describe('DatabaseService.getStackActivity', () => {
const ids: number[] = [];
for (let i = 0; i < 5; i++) {
const row = db.addNotificationHistory(0, { level: 'info', message: `e-${i}`, timestamp: ts, stack_name: 's' });
if (typeof row.id !== 'number') throw new Error('notification row missing id');
ids.push(row.id);
}
// ids[0..4] all share ts. Cursor at the third row's id should return ids[0] and ids[1].
const out = db.getStackActivity(0, 's', { limit: 50, before: ts, beforeId: ids[2] });
const cursorId = ids[2];
if (cursorId === undefined) throw new Error('cursor id missing');
const out = db.getStackActivity(0, 's', { limit: 50, before: ts, beforeId: cursorId });
const returnedIds = out.map((e: any) => e.id);
expect(returnedIds).toEqual([ids[1], ids[0]]);
});
@@ -154,6 +157,74 @@ describe('DatabaseService.getStackActivity', () => {
});
});
describe('DatabaseService.clearSelfContainerNotificationRouting', () => {
it('clears routing fields only from self Docker-event monitor notifications', () => {
const ts = Date.now();
const byProjectWithoutContainer = db.addNotificationHistory(0, {
level: 'error',
category: 'monitor_alert',
actor_username: 'system:docker-events',
message: 'sencho project crash without container',
timestamp: ts,
stack_name: 'sencho',
});
const byContainer = db.addNotificationHistory(0, {
level: 'error',
category: 'monitor_alert',
actor_username: 'system:docker-events',
message: 'sencho container crash',
timestamp: ts + 1,
stack_name: 'other',
container_name: 'sencho',
});
const sameProjectOtherContainer = db.addNotificationHistory(0, {
level: 'error',
category: 'monitor_alert',
actor_username: 'system:docker-events',
message: 'sencho sidecar crash',
timestamp: ts + 2,
stack_name: 'sencho',
container_name: 'sidecar',
});
const otherActor = db.addNotificationHistory(0, {
level: 'error',
category: 'monitor_alert',
actor_username: 'system:monitor',
message: 'user alert',
timestamp: ts + 3,
stack_name: 'sencho',
container_name: 'sencho',
});
const otherStack = db.addNotificationHistory(0, {
level: 'error',
category: 'monitor_alert',
actor_username: 'system:docker-events',
message: 'web crash',
timestamp: ts + 4,
stack_name: 'web',
container_name: 'web-1',
});
const changed = db.clearSelfContainerNotificationRouting(0, {
containerName: 'sencho',
composeProjectName: 'sencho',
});
expect(changed).toBe(2);
const rows = new Map(db.getNotificationHistory(0, 50).map(row => [row.message, row] as const));
expect(rows.get(byProjectWithoutContainer.message)?.stack_name).toBeUndefined();
expect(rows.get(byProjectWithoutContainer.message)?.container_name).toBeUndefined();
expect(rows.get(byContainer.message)?.stack_name).toBeUndefined();
expect(rows.get(byContainer.message)?.container_name).toBeUndefined();
expect(rows.get(sameProjectOtherContainer.message)?.stack_name).toBe('sencho');
expect(rows.get(sameProjectOtherContainer.message)?.container_name).toBe('sidecar');
expect(rows.get(otherActor.message)?.stack_name).toBe('sencho');
expect(rows.get(otherActor.message)?.container_name).toBe('sencho');
expect(rows.get(otherStack.message)?.stack_name).toBe('web');
expect(rows.get(otherStack.message)?.container_name).toBe('web-1');
});
});
describe('DatabaseService.addNotificationHistory (no per-insert prune)', () => {
it('keeps a quiet stack visible even after a chatty stack writes past the old 100-row per-node cap', () => {
const ts = Date.now();
+341 -18
View File
@@ -20,6 +20,7 @@ import request from 'supertest';
import bcrypt from 'bcrypt';
import { promises as fs } from 'fs';
import path from 'path';
import { Readable } from 'stream';
import { setupTestDb, cleanupTestDb, loginAsTestAdmin } from './helpers/setupTestDb';
// On Windows, fs.unlink on a directory returns EPERM instead of EISDIR so the
@@ -36,6 +37,56 @@ let viewerCookie: string;
let stacksDir: string;
const STACK = 'teststack';
type FileExplorerMetricEntry = {
op: string;
count: number;
successCount: number;
errorCount: number;
};
type TestWritableResponse = NodeJS.WritableStream & {
headersSent?: boolean;
req?: { emit: (event: 'close') => boolean };
destroy: (error?: Error) => void;
};
function asTestWritableResponse(destination: NodeJS.WritableStream): TestWritableResponse {
return destination as unknown as TestWritableResponse;
}
async function resetFileExplorerMetrics(): Promise<void> {
const { FileExplorerMetricsService } = await import('../services/FileExplorerMetricsService');
FileExplorerMetricsService.resetForTests();
}
async function getDownloadMetric(): Promise<FileExplorerMetricEntry | undefined> {
const metricsRes = await request(app)
.get('/api/file-explorer-metrics')
.set('Cookie', adminCookie);
return (metricsRes.body.entries as FileExplorerMetricEntry[]).find(
e => e.op === 'download',
);
}
async function expectDownloadMetricCounts(count: number, successCount: number, errorCount: number): Promise<void> {
let downloadEntry: FileExplorerMetricEntry | undefined;
for (let attempt = 0; attempt < 20; attempt++) {
downloadEntry = await getDownloadMetric();
if (
downloadEntry?.count === count &&
downloadEntry.successCount === successCount &&
downloadEntry.errorCount === errorCount
) {
return;
}
await new Promise<void>((resolve) => setTimeout(resolve, 10));
}
expect(downloadEntry).toBeDefined();
expect(downloadEntry!.count).toBe(count);
expect(downloadEntry!.successCount).toBe(successCount);
expect(downloadEntry!.errorCount).toBe(errorCount);
}
beforeAll(async () => {
tmpDir = await setupTestDb();
stacksDir = process.env.COMPOSE_DIR!;
@@ -355,6 +406,167 @@ describe('GET /api/stacks/:stackName/files/content', () => {
// ── GET /:stackName/files/download ────────────────────────────────────────────
describe('GET /api/stacks/:stackName/files/download', () => {
class CloseBeforeEndStream extends Readable {
public bytesRead: number;
constructor(private readonly payload: Buffer) {
super();
this.bytesRead = payload.length;
}
_read(): void {}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
destination.write(this.payload);
this.emit('close');
this.emit('end');
destination.end();
return destination;
}
}
class RequestCloseBeforeReadStream extends Readable {
public bytesRead = 0;
public destroyCalls = 0;
public emittedRequestClose = false;
constructor(private readonly payload: Buffer) {
super();
}
_read(): void {}
override destroy(error?: Error): this {
this.destroyCalls += 1;
return super.destroy(error);
}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
const req = asTestWritableResponse(destination).req;
if (req) {
this.emittedRequestClose = true;
req.emit('close');
}
destination.write(this.payload);
this.bytesRead = this.payload.length;
this.emit('close');
this.emit('end');
destination.end();
return destination;
}
}
class ResponseCloseBeforeReadStream extends Readable {
public bytesRead = 0;
public destroyCalls = 0;
public emittedResponseClose = false;
_read(): void {}
override destroy(_error?: Error): this {
this.destroyCalls += 1;
return this;
}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
this.emittedResponseClose = destination.emit('close');
asTestWritableResponse(destination).destroy(new Error('synthetic client abort'));
return destination;
}
}
class ResponseCloseThenFullReadStream extends Readable {
public bytesRead = 0;
public destroyCalls = 0;
public emittedResponseClose = false;
constructor(private readonly payload: Buffer) {
super();
}
_read(): void {}
override destroy(error?: Error): this {
this.destroyCalls += 1;
return super.destroy(error);
}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
this.emittedResponseClose = destination.emit('close');
destination.write(this.payload);
this.bytesRead = this.payload.length;
this.emit('end');
this.emit('close');
destination.end();
return destination;
}
}
class ErrorThenCloseStream extends Readable {
_read(): void {}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
this.emit('error', new Error('synthetic read failure'));
this.emit('close');
return destination;
}
}
class PrematureCloseOnlyStream extends Readable {
public bytesRead = 0;
_read(): void {}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
this.emit('close');
asTestWritableResponse(destination).destroy(new Error('synthetic premature close'));
return destination;
}
}
class EarlyEndBeforeFullReadStream extends Readable {
public bytesRead = 1;
_read(): void {}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
this.emit('end');
this.emit('close');
asTestWritableResponse(destination).destroy(new Error('synthetic early end'));
return destination;
}
}
class PartialWriteThenErrorStream extends Readable {
public headersWereSent = false;
constructor(private readonly payload: Buffer) {
super();
}
_read(): void {}
override pipe<T extends NodeJS.WritableStream>(destination: T): T {
destination.write(this.payload);
this.headersWereSent = Boolean(asTestWritableResponse(destination).headersSent);
this.emit('error', new Error('synthetic partial read failure'));
this.emit('close');
return destination;
}
}
async function mockDownloadStream(stream: Readable, size: number): Promise<void> {
const { FileSystemService } = await import('../services/FileSystemService');
vi.spyOn(FileSystemService, 'getInstance').mockReturnValue({
streamStackFile: vi.fn().mockResolvedValue({
stream,
size,
filename: 'compose.yaml',
mime: 'text/yaml',
}),
} as unknown as InstanceType<typeof FileSystemService>);
}
it('rejects unauthenticated requests with 401', async () => {
const res = await request(app)
.get(`/api/stacks/${STACK}/files/download`)
@@ -390,13 +602,10 @@ describe('GET /api/stacks/:stackName/files/download', () => {
expect(res.text).toContain('services');
});
it('records the download metric exactly once per successful response', async () => {
// The metric recorder is wired to both res.on("finish") and
// res.on("close"), guarded by a flag so a single completion does not
// double-fire. A regression that drops the flag would push successCount
// to 2 for one download.
const { FileExplorerMetricsService } = await import('../services/FileExplorerMetricsService');
FileExplorerMetricsService.resetForTests();
it('records the download metric exactly once per successful file read', async () => {
// The metric recorder is wired to the file stream lifecycle and guarded
// by a flag so a later stream event cannot record the same download again.
await resetFileExplorerMetrics();
const res = await request(app)
.get(`/api/stacks/${STACK}/files/download`)
@@ -404,19 +613,133 @@ describe('GET /api/stacks/:stackName/files/download', () => {
.set('Cookie', adminCookie);
expect(res.status).toBe(200);
// Allow the res.on('close') tail event to fire after the test's await.
await new Promise<void>((r) => setTimeout(r, 50));
await expectDownloadMetricCounts(1, 1, 0);
});
const metricsRes = await request(app)
.get('/api/file-explorer-metrics')
it('records a fully read stream as success when close fires before end', async () => {
await resetFileExplorerMetrics();
const payload = Buffer.from('services: {}\n');
await mockDownloadStream(new CloseBeforeEndStream(payload), payload.length);
const res = await request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie);
const downloadEntry = (metricsRes.body.entries as Array<{ op: string; count: number; successCount: number; errorCount: number }>).find(
e => e.op === 'download',
);
expect(downloadEntry).toBeDefined();
expect(downloadEntry!.count).toBe(1);
expect(downloadEntry!.successCount).toBe(1);
expect(downloadEntry!.errorCount).toBe(0);
expect(res.status).toBe(200);
expect(res.text).toBe(payload.toString('utf-8'));
await expectDownloadMetricCounts(1, 1, 0);
});
it('does not destroy the stream when request close fires before the file is read', async () => {
await resetFileExplorerMetrics();
const payload = Buffer.from('services: {}\n');
const stream = new RequestCloseBeforeReadStream(payload);
await mockDownloadStream(stream, payload.length);
const res = await request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie);
expect(res.status).toBe(200);
expect(res.text).toBe(payload.toString('utf-8'));
expect(stream.emittedRequestClose).toBe(true);
expect(stream.destroyCalls).toBe(0);
await expectDownloadMetricCounts(1, 1, 0);
});
it('records response abort cleanup exactly once as a failed download', async () => {
await resetFileExplorerMetrics();
const stream = new ResponseCloseBeforeReadStream();
await mockDownloadStream(stream, 16);
await expect(request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie)).rejects.toThrow();
expect(stream.emittedResponseClose).toBe(true);
await expectDownloadMetricCounts(1, 0, 1);
expect(stream.destroyCalls).toBe(1);
});
it('lets full source completion win when response close fires first', async () => {
await resetFileExplorerMetrics();
const payload = Buffer.from('services: {}\n');
const stream = new ResponseCloseThenFullReadStream(payload);
await mockDownloadStream(stream, payload.length);
const res = await request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie);
expect(res.status).toBe(200);
expect(stream.emittedResponseClose).toBe(true);
expect(stream.destroyCalls).toBe(0);
await expectDownloadMetricCounts(1, 1, 0);
});
it('records stream error followed by close exactly once as a failed download', async () => {
await resetFileExplorerMetrics();
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined);
await mockDownloadStream(new ErrorThenCloseStream(), 16);
try {
const res = await request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie);
expect(res.status).toBe(500);
} finally {
errorSpy.mockRestore();
}
await expectDownloadMetricCounts(1, 0, 1);
});
it('records bare source close before full read exactly once as a failed download', async () => {
await resetFileExplorerMetrics();
await mockDownloadStream(new PrematureCloseOnlyStream(), 16);
await expect(request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie)).rejects.toThrow();
await expectDownloadMetricCounts(1, 0, 1);
});
it('records source end before full read exactly once as a failed download', async () => {
await resetFileExplorerMetrics();
await mockDownloadStream(new EarlyEndBeforeFullReadStream(), 16);
await expect(request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie)).rejects.toThrow();
await expectDownloadMetricCounts(1, 0, 1);
});
it('records a partial stream error after headers are sent exactly once', async () => {
await resetFileExplorerMetrics();
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined);
const stream = new PartialWriteThenErrorStream(Buffer.from('partial'));
await mockDownloadStream(stream, 16);
try {
await expect(request(app)
.get(`/api/stacks/${STACK}/files/download`)
.query({ path: 'compose.yaml' })
.set('Cookie', adminCookie)).rejects.toThrow();
} finally {
errorSpy.mockRestore();
}
expect(stream.headersWereSent).toBe(true);
await expectDownloadMetricCounts(1, 0, 1);
});
});