From 265fece988804bf2b983080dd45331963df5981f Mon Sep 17 00:00:00 2001 From: Anso Date: Thu, 28 May 2026 12:45:10 -0400 Subject: [PATCH] fix(notifications): prevent self-container stack routing (#1242) * fix(notifications): prevent self-container stack routing * fix(stack-files): stabilize download metrics in CI --- .../__tests__/docker-event-service.test.ts | 118 ++++++ .../__tests__/self-identity-service.test.ts | 28 ++ backend/src/__tests__/stack-activity.test.ts | 75 +++- .../src/__tests__/stack-files-routes.test.ts | 359 +++++++++++++++++- backend/src/bootstrap/startup.ts | 26 +- backend/src/routes/stacks.ts | 55 ++- backend/src/services/DatabaseService.ts | 32 ++ backend/src/services/DockerEventService.ts | 48 ++- backend/src/services/SelfIdentityService.ts | 16 +- 9 files changed, 704 insertions(+), 53 deletions(-) diff --git a/backend/src/__tests__/docker-event-service.test.ts b/backend/src/__tests__/docker-event-service.test.ts index bad8199e..e405ee80 100644 --- a/backend/src/__tests__/docker-event-service.test.ts +++ b/backend/src/__tests__/docker-event-service.test.ts @@ -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(); diff --git a/backend/src/__tests__/self-identity-service.test.ts b/backend/src/__tests__/self-identity-service.test.ts index e11c4b2f..81676eac 100644 --- a/backend/src/__tests__/self-identity-service.test.ts +++ b/backend/src/__tests__/self-identity-service.test.ts @@ -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(); diff --git a/backend/src/__tests__/stack-activity.test.ts b/backend/src/__tests__/stack-activity.test.ts index 163fd8d4..47967dbb 100644 --- a/backend/src/__tests__/stack-activity.test.ts +++ b/backend/src/__tests__/stack-activity.test.ts @@ -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(); diff --git a/backend/src/__tests__/stack-files-routes.test.ts b/backend/src/__tests__/stack-files-routes.test.ts index e5460bfc..43b09f79 100644 --- a/backend/src/__tests__/stack-files-routes.test.ts +++ b/backend/src/__tests__/stack-files-routes.test.ts @@ -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 { + const { FileExplorerMetricsService } = await import('../services/FileExplorerMetricsService'); + FileExplorerMetricsService.resetForTests(); +} + +async function getDownloadMetric(): Promise { + 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 { + 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((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(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(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(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(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(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(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(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(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 { + 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); + } + 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((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); }); }); diff --git a/backend/src/bootstrap/startup.ts b/backend/src/bootstrap/startup.ts index c98b672f..3af518c3 100644 --- a/backend/src/bootstrap/startup.ts +++ b/backend/src/bootstrap/startup.ts @@ -52,6 +52,20 @@ export function ensurePilotJwtSecret(): boolean { return true; } +function clearSelfContainerNotificationRouting(): void { + const identity = SelfIdentityService.getInstance().getIdentity(); + const changed = DatabaseService.getInstance().clearSelfContainerNotificationRouting( + NodeRegistry.getInstance().getDefaultNodeId(), + { + containerName: identity.containerName, + composeProjectName: identity.composeProjectName, + }, + ); + if (changed > 0) { + console.log(`[Startup] Cleared stack routing from ${changed} Sencho self-container notification(s)`); + } +} + /** * Run the startup sequence: stack-directory migration, service initialization, * background watchdogs, then bind the HTTP server. The caller passes the @@ -118,12 +132,16 @@ export async function startServer(server: Server): Promise { // the live loopback bridge instead of waiting for the 3-minute TTL. PilotTunnelManager.getInstance().on('tunnel-up', invalidateRemoteMetaCache); - // Async initializers are independent of each other; run in parallel - // so total boot time is the slowest one rather than the sum. + // Most async initializers still run in parallel. Docker event monitoring + // is sequenced after self identity so it never classifies Sencho's own + // container as a routeable stack event. await Promise.all([ SelfUpdateService.getInstance().initialize(), - SelfIdentityService.getInstance().initialize(), - DockerEventManager.getInstance().start(), + (async () => { + await SelfIdentityService.getInstance().initialize(); + clearSelfContainerNotificationRouting(); + await DockerEventManager.getInstance().start(); + })(), TrivyService.getInstance().initialize(), ]); diff --git a/backend/src/routes/stacks.ts b/backend/src/routes/stacks.ts index 86aa4d44..4a1cbe15 100644 --- a/backend/src/routes/stacks.ts +++ b/backend/src/routes/stacks.ts @@ -1459,33 +1459,54 @@ stacksRouter.get('/:stackName/files/download', async (req: Request, res: Respons const encodedFilename = encodeURIComponent(result.filename); const safeFilename = result.filename.replace(/[\\"]/g, ''); res.setHeader('Content-Disposition', `attachment; filename="${safeFilename}"; filename*=UTF-8''${encodedFilename}`); - // Track download completion off the file stream's lifecycle, not the - // response's. Node's `res.on('finish')` and `res.on('close')` fire in - // platform-dependent order under the in-process supertest transport - // (close occasionally precedes finish, which made the success-vs-error - // recorder racy in CI). The file stream's `end`, `error`, and `close` - // events ARE deterministic: a clean read emits `end` then `close`; a - // destroy emits `close` without `end`; a disk error emits `error` then - // `close`. Hanging the recorder off these signals removes the race. + // Track normal download completion off the file stream's lifecycle, not + // the request lifecycle. Under the in-process supertest transport, request + // close events can race ahead of normal stream completion; response close + // is handled below only as delayed abort cleanup. + // End/error are the durable completion signals, and close can still count + // as success only when the stream reports it read the full file. let downloadRecorded = false; + const streamWithBytes = result.stream as typeof result.stream & { bytesRead?: number }; + const hasReadFullFile = (): boolean => ( + result.size === 0 || + (typeof streamWithBytes.bytesRead === 'number' && streamWithBytes.bytesRead >= result.size) + ); + let abortCleanupHandle: NodeJS.Immediate | null = null; + const clearAbortCleanup = (): void => { + if (!abortCleanupHandle) return; + clearImmediate(abortCleanupHandle); + abortCleanupHandle = null; + }; const recordDownloadOnce = (ok: boolean): void => { if (downloadRecorded) return; downloadRecorded = true; + clearAbortCleanup(); recordFileOp(req.nodeId, 'download', startedAt, ok); }; result.stream.on('error', (streamErr) => { console.error('[files] stream error:', sanitizeForLog(getErrorMessage(streamErr, 'unknown'))); - if (!res.headersSent) res.status(500).end(); - else res.destroy(); + if (!res.headersSent) { + res.removeHeader('Content-Length'); + res.status(500).end(); + } else { + res.destroy(); + } recordDownloadOnce(false); }); - result.stream.on('end', () => recordDownloadOnce(true)); - result.stream.on('close', () => { - if (res.writableEnded) recordDownloadOnce(true); - else recordDownloadOnce(false); - }); - req.on('close', () => { - if (!res.writableEnded) result.stream.destroy(); + result.stream.on('end', () => recordDownloadOnce(hasReadFullFile())); + result.stream.on('close', () => recordDownloadOnce(hasReadFullFile())); + res.on('close', () => { + if (downloadRecorded || hasReadFullFile()) return; + // At this point, response close is treated as an abort signal. It can + // beat the source stream's final events in the in-process test transport. + // Give a same-turn clean source completion a chance to win before cleanup. + abortCleanupHandle = setImmediate(() => { + abortCleanupHandle = null; + if (downloadRecorded || hasReadFullFile()) return; + recordDownloadOnce(false); + result.stream.destroy(); + }); + abortCleanupHandle.unref?.(); }); logFileDiag('download stream opened', { stackName, relPath, nodeId: req.nodeId, size: result.size, elapsedMs: Date.now() - startedAt }); result.stream.pipe(res); diff --git a/backend/src/services/DatabaseService.ts b/backend/src/services/DatabaseService.ts index e4714815..2567769c 100644 --- a/backend/src/services/DatabaseService.ts +++ b/backend/src/services/DatabaseService.ts @@ -2046,6 +2046,38 @@ export class DatabaseService { }; } + public clearSelfContainerNotificationRouting( + nodeId: number, + self: { containerName?: string | null; composeProjectName?: string | null }, + ): number { + const containerName = self.containerName?.trim() || null; + const composeProjectName = self.composeProjectName?.trim() || null; + if (!containerName && !composeProjectName) return 0; + + const predicates: string[] = []; + const args: (number | string)[] = [nodeId]; + if (containerName) { + predicates.push('container_name = ?'); + args.push(containerName); + } + if (composeProjectName) { + predicates.push('(container_name IS NULL AND stack_name = ?)'); + args.push(composeProjectName); + } + + const result = this.db.prepare(` + UPDATE notification_history + SET stack_name = NULL, + container_name = NULL + WHERE node_id = ? + AND actor_username = 'system:docker-events' + AND category = 'monitor_alert' + AND (stack_name IS NOT NULL OR container_name IS NOT NULL) + AND (${predicates.join(' OR ')}) + `).run(...args); + return result.changes; + } + public getStackActivity(nodeId: number, stackName: string, opts: { limit: number; before?: number; beforeId?: number }): NotificationHistory[] { // Composite (timestamp, id) cursor: pure timestamp pagination drops rows // on same-millisecond bursts (Docker events from one compose up). diff --git a/backend/src/services/DockerEventService.ts b/backend/src/services/DockerEventService.ts index 0b9af8d8..81f74cff 100644 --- a/backend/src/services/DockerEventService.ts +++ b/backend/src/services/DockerEventService.ts @@ -2,6 +2,7 @@ import Docker from 'dockerode'; import { NodeRegistry } from './NodeRegistry'; import { NotificationCategory, NotificationService } from './NotificationService'; import { DatabaseService } from './DatabaseService'; +import SelfIdentityService from './SelfIdentityService'; import { classifyDie, classifyGapExit, @@ -79,6 +80,7 @@ const SETTINGS_CACHE_MS = 500; interface InternalContainerState extends ContainerLifecycleState { name?: string; stackName?: string; + isSelf?: boolean; lastCrashAlertAt?: number; lastActivityAt: number; healthStatus?: 'healthy' | 'unhealthy' | 'starting'; @@ -341,9 +343,10 @@ export class DockerEventService { const name = inspect.Name?.replace(/^\//, '') ?? containerId.slice(0, 12); const stackName = inspect.Config?.Labels?.[COMPOSE_PROJECT_LABEL]; const exitCode = inspect.State?.ExitCode ?? 0; + const isSelf = this.isSelfContainer(containerId, name); // Gap exits have no in-memory state, so there's no dedup to bump. - await this.emitClassification(classification, null, { name, exitCode, stackName }); + await this.emitClassification(classification, null, { name, exitCode, stackName, isSelf }); } catch (err) { if (isDebugEnabled()) { console.log(`[DockerEvents:${this.nodeName}:diag] gap inspect failed:`, @@ -379,6 +382,8 @@ export class DockerEventService { const action = event.Action ?? ''; const id = event.Actor?.ID; if (!id) return; + const attrs = event.Actor?.Attributes ?? {}; + const isSelf = this.isSelfContainer(id, attrs.name); // Normalize: `health_status: unhealthy` -> base action const baseAction = action.startsWith('health_status') ? 'health_status' : action; @@ -387,12 +392,12 @@ export class DockerEventService { // refetch stack statuses immediately on a real container event, // without waiting for the next polling tick. This is fire-and-forget // and is NOT persisted to the alerts history. - if (STATE_INVALIDATE_ACTIONS.has(baseAction)) { + if (STATE_INVALIDATE_ACTIONS.has(baseAction) && !isSelf) { this.notifier.broadcastEvent({ type: 'state-invalidate', scope: 'stack', nodeId: this.nodeId, - stackName: event.Actor?.Attributes?.[COMPOSE_PROJECT_LABEL] ?? null, + stackName: attrs[COMPOSE_PROJECT_LABEL] ?? null, containerId: id, action: baseAction, ts: Date.now(), @@ -449,12 +454,12 @@ export class DockerEventService { state.healthStatus = 'unhealthy'; if (!this.isCrashAlertsEnabled()) return; const name = state.name ?? id.slice(0, 12); - const stackName = state.stackName; void this.emitError( 'monitor_alert', `Healthcheck failed: ${name} is unhealthy.`, - stackName, + state.stackName, state.name, + state.isSelf === true, ); } else { state.unhealthySince = undefined; @@ -534,6 +539,7 @@ export class DockerEventService { name: state.name ?? id.slice(0, 12), exitCode: exitCode ?? 0, stackName: state.stackName, + isSelf: state.isSelf === true, }); } @@ -544,7 +550,7 @@ export class DockerEventService { private async emitClassification( classification: Classification, state: InternalContainerState | null, - info: { name: string; exitCode: number; stackName?: string }, + info: { name: string; exitCode: number; stackName?: string; isSelf?: boolean }, ): Promise { // Respect the existing global crash-alerts toggle so users who have // disabled these notifications in Settings remain opted out. @@ -564,7 +570,7 @@ export class DockerEventService { // rate-suppressed alerts don't silently lock out the next real crash. if (state) state.lastCrashAlertAt = Date.now(); - await this.emitError('monitor_alert', message, info.stackName, info.name); + await this.emitError('monitor_alert', message, info.stackName, info.name, info.isSelf === true); } private isCrashAlertsEnabled(): boolean { @@ -649,9 +655,19 @@ export class DockerEventService { const project = attrs[COMPOSE_PROJECT_LABEL]; if (project && !state.stackName) state.stackName = project; } + if (this.isSelfContainer(id, state.name)) { + state.isSelf = true; + } return state; } + private isSelfContainer(id?: string, name?: string): boolean { + const self = SelfIdentityService.getInstance(); + if (id && self.isOwnContainer(id)) return true; + const normalizedName = name?.replace(/^\//, ''); + return normalizedName ? self.isOwnContainer(normalizedName) : false; + } + private eventTimeMs(event: DockerEventPayload): number { if (typeof event.timeNano === 'number') return Math.floor(event.timeNano / 1_000_000); if (typeof event.time === 'number') return event.time * 1000; @@ -672,16 +688,26 @@ export class DockerEventService { // Notification wrappers (prefix with node name for multi-node clarity) // ======================================================================== - private async emitError(category: NotificationCategory, message: string, stackName?: string, containerName?: string): Promise { - return this.notifier.dispatchAlert('error', category, this.prefix(message), { stackName, containerName, actor: 'system:docker-events' }); + private buildAlertOptions( + stackName?: string, + containerName?: string, + systemOnly = false, + ): { stackName?: string; containerName?: string; actor: string } { + return systemOnly + ? { actor: 'system:docker-events' } + : { stackName, containerName, actor: 'system:docker-events' }; + } + + private async emitError(category: NotificationCategory, message: string, stackName?: string, containerName?: string, systemOnly = false): Promise { + return this.notifier.dispatchAlert('error', category, this.prefix(message), this.buildAlertOptions(stackName, containerName, systemOnly)); } private async emitWarning(category: NotificationCategory, message: string, stackName?: string, containerName?: string): Promise { - return this.notifier.dispatchAlert('warning', category, this.prefix(message), { stackName, containerName, actor: 'system:docker-events' }); + return this.notifier.dispatchAlert('warning', category, this.prefix(message), this.buildAlertOptions(stackName, containerName)); } private async emitInfo(category: NotificationCategory, message: string, stackName?: string, containerName?: string): Promise { - return this.notifier.dispatchAlert('info', category, this.prefix(message), { stackName, containerName, actor: 'system:docker-events' }); + return this.notifier.dispatchAlert('info', category, this.prefix(message), this.buildAlertOptions(stackName, containerName)); } private prefix(message: string): string { diff --git a/backend/src/services/SelfIdentityService.ts b/backend/src/services/SelfIdentityService.ts index 429198f6..db04bb62 100644 --- a/backend/src/services/SelfIdentityService.ts +++ b/backend/src/services/SelfIdentityService.ts @@ -24,11 +24,13 @@ class SelfIdentityService { private static instance: SelfIdentityService; private containerId: string | null = null; private containerName: string | null = null; + private composeProjectName: string | null = null; private imageIdHex: string | null = null; private networkIds = new Set(); private networkNames = new Set(); private volumeNames = new Set(); private initialized = false; + private initializePromise: Promise | null = null; public static getInstance(): SelfIdentityService { if (!SelfIdentityService.instance) { @@ -38,15 +40,23 @@ class SelfIdentityService { } async initialize(): Promise { + if (this.initializePromise) return this.initializePromise; if (this.initialized) return; - this.initialized = true; + this.initializePromise = this.initializeInternal().finally(() => { + this.initialized = true; + this.initializePromise = null; + }); + return this.initializePromise; + } + private async initializeInternal(): Promise { const docker = DockerController.getInstance().getDocker(); const info = await this.resolveSelfInspect(docker); if (!info) return; this.containerId = info.Id ?? null; this.containerName = (info.Name || '').replace(/^\//, '') || null; + this.composeProjectName = info.Config?.Labels?.['com.docker.compose.project'] ?? null; this.imageIdHex = SelfIdentityService.stripSha(info.Image ?? '') || null; const nets = info.NetworkSettings?.Networks ?? {}; @@ -148,6 +158,7 @@ class SelfIdentityService { getIdentity(): { containerId: string | null; containerName: string | null; + composeProjectName: string | null; imageId: string | null; networkNames: string[]; volumeNames: string[]; @@ -155,6 +166,7 @@ class SelfIdentityService { return { containerId: this.containerId, containerName: this.containerName, + composeProjectName: this.composeProjectName, imageId: this.imageIdHex, networkNames: [...this.networkNames], volumeNames: [...this.volumeNames], @@ -165,11 +177,13 @@ class SelfIdentityService { resetForTesting(): void { this.containerId = null; this.containerName = null; + this.composeProjectName = null; this.imageIdHex = null; this.networkIds.clear(); this.networkNames.clear(); this.volumeNames.clear(); this.initialized = false; + this.initializePromise = null; } private static stripSha(s: string): string {