From 61ad3b6fbf1aa1a929220e7ad23d736d1b185de8 Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sat, 24 Oct 2026 22:26:13 -0400 Subject: [PATCH 1/8] fix(graphql): preserve partial results for unavailable services --- api/generated-schema.graphql | 4 +- api/src/unraid-api/cli/generated/graphql.ts | 4 +- .../unraid-api/graph/partial-results.spec.ts | 141 ++++++++++++++++++ .../graph/resolvers/docker/docker.resolver.ts | 2 +- .../graph/resolvers/vms/vms.resolver.ts | 2 +- web/src/composables/gql/graphql.ts | 12 +- 6 files changed, 153 insertions(+), 12 deletions(-) create mode 100644 api/src/unraid-api/graph/partial-results.spec.ts diff --git a/api/generated-schema.graphql b/api/generated-schema.graphql index b373873c11..0201eb59c5 100644 --- a/api/generated-schema.graphql +++ b/api/generated-schema.graphql @@ -3314,7 +3314,7 @@ type Query { vars: Vars! """Get information about all VMs on the system""" - vms: Vms! + vms: Vms parityHistory: [ParityCheck!]! array: UnraidArray! customization: Customization @@ -3326,7 +3326,7 @@ type Query { """Network interfaces""" networkInterfaces: [InfoNetworkInterface!]! - docker: Docker! + docker: Docker disks: [Disk!]! assignableDisks: [Disk!]! disk(id: PrefixedID!): Disk! diff --git a/api/src/unraid-api/cli/generated/graphql.ts b/api/src/unraid-api/cli/generated/graphql.ts index 5119a999ad..777ffdc7cc 100644 --- a/api/src/unraid-api/cli/generated/graphql.ts +++ b/api/src/unraid-api/cli/generated/graphql.ts @@ -2383,7 +2383,7 @@ export type Query = { disk: Disk; disks: Array; display: InfoDisplay; - docker: Docker; + docker?: Maybe; flash: Flash; /** Get JSON Schema for API key creation form */ getApiKeyCreationFormSchema: ApiKeyFormSettings; @@ -2447,7 +2447,7 @@ export type Query = { validateOidcSession: OidcSessionValidation; vars: Vars; /** Get information about all VMs on the system */ - vms: Vms; + vms?: Maybe; }; diff --git a/api/src/unraid-api/graph/partial-results.spec.ts b/api/src/unraid-api/graph/partial-results.spec.ts new file mode 100644 index 0000000000..cf80f5dcdc --- /dev/null +++ b/api/src/unraid-api/graph/partial-results.spec.ts @@ -0,0 +1,141 @@ +import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo'; +import { GraphQLModule, GraphQLSchemaHost, Query, Resolver } from '@nestjs/graphql'; +import { FastifyAdapter, NestFastifyApplication } from '@nestjs/platform-fastify'; +import { Test } from '@nestjs/testing'; +import { readFile } from 'fs/promises'; + +import { PrefixedID } from '@unraid/shared/prefixed-id-scalar.js'; +import { buildSchema } from 'graphql'; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { DockerResolver } from '@app/unraid-api/graph/resolvers/docker/docker.resolver.js'; +import { DockerService } from '@app/unraid-api/graph/resolvers/docker/docker.service.js'; +import { VmsResolver } from '@app/unraid-api/graph/resolvers/vms/vms.resolver.js'; +import { VmsService } from '@app/unraid-api/graph/resolvers/vms/vms.service.js'; +import { SubscriptionTrackerService } from '@app/unraid-api/graph/services/subscription-tracker.service.js'; + +@Resolver() +class ServerStatusResolver { + @Query(() => String) + serverStatus() { + return 'available'; + } +} + +type Result = { + data: Record | null; + errors?: { path: (string | number)[] }[]; +}; + +describe('Partial GraphQL results for unavailable services', () => { + let app: NestFastifyApplication; + const vms = { getDomains: vi.fn() }; + const docker = { getNetworks: vi.fn() }; + + beforeAll(async () => { + const module = await Test.createTestingModule({ + imports: [ + GraphQLModule.forRoot({ + driver: ApolloDriver, + autoSchemaFile: true, + }), + ], + providers: [ + PrefixedID, + ServerStatusResolver, + VmsResolver, + DockerResolver, + { provide: VmsService, useValue: vms }, + { provide: DockerService, useValue: docker }, + { + provide: SubscriptionTrackerService, + useValue: { registerTopic: vi.fn() }, + }, + ], + }) + .useMocker(() => ({})) + .compile(); + app = module.createNestApplication(new FastifyAdapter()); + await app.init(); + await app.getHttpAdapter().getInstance().ready(); + }); + + beforeEach(() => { + vms.getDomains.mockReset().mockResolvedValue([]); + docker.getNetworks.mockReset().mockResolvedValue([]); + }); + + afterAll(async () => app?.close()); + + async function execute(query: string): Promise { + const response = await app.inject({ method: 'POST', url: '/graphql', payload: { query } }); + expect(response.statusCode).toBe(200); + return response.json(); + } + + it('keeps the runtime and published service roots nullable', async () => { + const published = buildSchema(await readFile('generated-schema.graphql', 'utf8')); + const runtime = app.get(GraphQLSchemaHost).schema; + for (const schema of [published, runtime]) { + for (const root of ['vms', 'docker']) { + expect(schema.getQueryType()!.getFields()[root].type.toString()).toBe( + root === 'vms' ? 'Vms' : 'Docker' + ); + } + } + }); + + it.each(['domain', 'domains'])('preserves other roots when vms.%s fails', async (field) => { + vms.getDomains.mockRejectedValue(new Error('VMs are not available')); + const result = await execute( + `{ serverStatus vms { ${field} { id } } docker { networks { id } } }` + ); + expect(result.data).toEqual({ + serverStatus: 'available', + vms: { [field]: null }, + docker: { networks: [] }, + }); + expect(result.errors?.map((error) => error.path)).toEqual([['vms', field]]); + }); + + it('preserves other roots when Docker fails', async () => { + docker.getNetworks.mockRejectedValue(new Error('Docker is not available')); + const result = await execute( + '{ serverStatus vms { domains { id } } docker { networks { id } } }' + ); + expect(result.data).toEqual({ serverStatus: 'available', vms: { domains: [] }, docker: null }); + expect(result.errors?.map((error) => error.path)).toEqual([['docker', 'networks']]); + }); + + it('retains successful data and both errors when both services fail', async () => { + vms.getDomains.mockRejectedValue(new Error('VMs are not available')); + docker.getNetworks.mockRejectedValue(new Error('Docker is not available')); + const result = await execute( + '{ serverStatus guests: vms { domains { id } } containers: docker { networks { id } } }' + ); + expect(result.data).toEqual({ + serverStatus: 'available', + guests: { domains: null }, + containers: null, + }); + expect(result.errors?.map((error) => error.path)).toEqual( + expect.arrayContaining([ + ['guests', 'domains'], + ['containers', 'networks'], + ]) + ); + expect(result.errors).toHaveLength(2); + }); + + it('preserves successful empty service results', async () => { + const result = await execute( + '{ serverStatus vms { domains { id } } docker { networks { id } } }' + ); + expect(result.errors).toBeUndefined(); + expect(result.data).toEqual({ + serverStatus: 'available', + vms: { domains: [] }, + docker: { networks: [] }, + }); + }); +}); diff --git a/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts b/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts index e8d025c5f8..cc67450dc8 100644 --- a/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts +++ b/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts @@ -64,7 +64,7 @@ export class DockerResolver { action: AuthAction.READ_ANY, resource: Resource.DOCKER, }) - @Query(() => Docker) + @Query(() => Docker, { nullable: true }) public docker() { return { id: 'docker', diff --git a/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts b/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts index 13f066ce7f..481d6fdf7d 100644 --- a/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts +++ b/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts @@ -10,7 +10,7 @@ import { VmsService } from '@app/unraid-api/graph/resolvers/vms/vms.service.js'; export class VmsResolver { constructor(private readonly vmsService: VmsService) {} - @Query(() => Vms, { description: 'Get information about all VMs on the system' }) + @Query(() => Vms, { nullable: true, description: 'Get information about all VMs on the system' }) @UsePermissions({ action: AuthAction.READ_ANY, resource: Resource.VMS, diff --git a/web/src/composables/gql/graphql.ts b/web/src/composables/gql/graphql.ts index e9be3b367b..ff17bb25e5 100644 --- a/web/src/composables/gql/graphql.ts +++ b/web/src/composables/gql/graphql.ts @@ -2390,7 +2390,7 @@ export type Query = { disk: Disk; disks: Array; display: InfoDisplay; - docker: Docker; + docker?: Maybe; flash: Flash; /** Get JSON Schema for API key creation form */ getApiKeyCreationFormSchema: ApiKeyFormSettings; @@ -2454,7 +2454,7 @@ export type Query = { validateOidcSession: OidcSessionValidation; vars: Vars; /** Get information about all VMs on the system */ - vms: Vms; + vms?: Maybe; }; @@ -3690,14 +3690,14 @@ export type SetThemeMutation = { __typename?: 'Mutation', customization: { __typ export type GetDockerContainerSizesQueryVariables = Exact<{ [key: string]: never; }>; -export type GetDockerContainerSizesQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', id: string, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, sizeRootFs?: number | null, sizeRw?: number | null, sizeLog?: number | null }> } }; +export type GetDockerContainerSizesQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', id: string, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, sizeRootFs?: number | null, sizeRw?: number | null, sizeLog?: number | null }> } | null }; export type GetDockerContainersQueryVariables = Exact<{ skipCache?: InputMaybe; }>; -export type GetDockerContainersQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', id: string, portConflicts: { __typename?: 'DockerPortConflicts', containerPorts: Array<{ __typename?: 'DockerContainerPortConflict', privatePort: number, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }>, lanPorts: Array<{ __typename?: 'DockerLanPortConflict', lanIpPort: string, publicPort?: number | null, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }> }, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, created: number, lanIpPorts?: Array | null, autoStart: boolean, autoStartOrder?: number | null, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null }>, organizer: { __typename?: 'ResolvedOrganizerV1', version: number, views: Array<{ __typename?: 'ResolvedOrganizerView', id: string, name: string, rootId: string, prefs?: any | null, flatEntries: Array<{ __typename?: 'FlatOrganizerEntry', id: string, type: string, name: string, parentId?: string | null, depth: number, position: number, path: Array, hasChildren: boolean, childrenIds: Array, meta?: { __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, lanIpPorts?: Array | null, autoStart: boolean, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, created: number, isUpdateAvailable?: boolean | null, isRebuildReady?: boolean | null, templatePath?: string | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null } | null }> }> } } }; +export type GetDockerContainersQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', id: string, portConflicts: { __typename?: 'DockerPortConflicts', containerPorts: Array<{ __typename?: 'DockerContainerPortConflict', privatePort: number, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }>, lanPorts: Array<{ __typename?: 'DockerLanPortConflict', lanIpPort: string, publicPort?: number | null, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }> }, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, created: number, lanIpPorts?: Array | null, autoStart: boolean, autoStartOrder?: number | null, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null }>, organizer: { __typename?: 'ResolvedOrganizerV1', version: number, views: Array<{ __typename?: 'ResolvedOrganizerView', id: string, name: string, rootId: string, prefs?: any | null, flatEntries: Array<{ __typename?: 'FlatOrganizerEntry', id: string, type: string, name: string, parentId?: string | null, depth: number, position: number, path: Array, hasChildren: boolean, childrenIds: Array, meta?: { __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, lanIpPorts?: Array | null, autoStart: boolean, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, created: number, isUpdateAvailable?: boolean | null, isRebuildReady?: boolean | null, templatePath?: string | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null } | null }> }> } } | null }; export type CreateDockerFolderWithItemsMutationVariables = Exact<{ name: Scalars['String']['input']; @@ -3732,7 +3732,7 @@ export type GetDockerContainerLogsQueryVariables = Exact<{ }>; -export type GetDockerContainerLogsQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', logs: { __typename?: 'DockerContainerLogs', containerId: string, cursor?: string | null, lines: Array<{ __typename?: 'DockerContainerLogLine', timestamp: string, message: string }> } } }; +export type GetDockerContainerLogsQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', logs: { __typename?: 'DockerContainerLogs', containerId: string, cursor?: string | null, lines: Array<{ __typename?: 'DockerContainerLogLine', timestamp: string, message: string }> } } | null }; export type MoveDockerEntriesToFolderMutationVariables = Exact<{ destinationFolderId: Scalars['String']['input']; @@ -3808,7 +3808,7 @@ export type GetContainerTailscaleStatusQueryVariables = Exact<{ }>; -export type GetContainerTailscaleStatusQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', container?: { __typename?: 'DockerContainer', id: string, tailscaleStatus?: { __typename?: 'TailscaleStatus', online: boolean, version?: string | null, latestVersion?: string | null, updateAvailable: boolean, hostname?: string | null, dnsName?: string | null, relay?: string | null, relayName?: string | null, tailscaleIps?: Array | null, primaryRoutes?: Array | null, isExitNode: boolean, webUiUrl?: string | null, keyExpiry?: string | null, keyExpiryDays?: number | null, keyExpired: boolean, backendState?: string | null, authUrl?: string | null, exitNodeStatus?: { __typename?: 'TailscaleExitNodeStatus', online: boolean, tailscaleIps?: Array | null } | null } | null } | null } }; +export type GetContainerTailscaleStatusQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', container?: { __typename?: 'DockerContainer', id: string, tailscaleStatus?: { __typename?: 'TailscaleStatus', online: boolean, version?: string | null, latestVersion?: string | null, updateAvailable: boolean, hostname?: string | null, dnsName?: string | null, relay?: string | null, relayName?: string | null, tailscaleIps?: Array | null, primaryRoutes?: Array | null, isExitNode: boolean, webUiUrl?: string | null, keyExpiry?: string | null, keyExpiryDays?: number | null, keyExpired: boolean, backendState?: string | null, authUrl?: string | null, exitNodeStatus?: { __typename?: 'TailscaleExitNodeStatus', online: boolean, tailscaleIps?: Array | null } | null } | null } | null } | null }; export type UnpauseDockerContainerMutationVariables = Exact<{ id: Scalars['PrefixedID']['input']; From 894b88ca3533d0b7039f6deb1d0bdfb48fa36b24 Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sat, 24 Oct 2026 22:27:35 -0400 Subject: [PATCH 2/8] fix(notifications): count only visible archived notifications --- .../notifications.overview.spec.ts | 207 ++++++++++++++++++ .../notifications/notifications.service.ts | 181 ++++++--------- 2 files changed, 268 insertions(+), 120 deletions(-) create mode 100644 api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts new file mode 100644 index 0000000000..2ea664c78c --- /dev/null +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts @@ -0,0 +1,207 @@ +import { ConfigService } from '@nestjs/config'; +import { mkdir, mkdtemp, rm, unlink, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import type { FSWatcher } from 'chokidar'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { NotificationType } from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; + +describe.sequential('Notification overview visibility', () => { + let basePath: string; + let service: NotificationsService; + const file = (type: NotificationType, id: string) => join(basePath, type.toLowerCase(), id); + + async function write(type: NotificationType, id: string, importance = 'normal') { + await writeFile( + file(type, id), + `timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=${importance}\n` + ); + } + + async function start(keepWatcher = false) { + service = new NotificationsService( + new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }) + ); + await Reflect.get(service, 'initialization'); + if (!keepWatcher) await stopWatcher(); + } + + async function stopWatcher() { + const watcher: FSWatcher | null = Reflect.get(NotificationsService, 'watcher'); + await watcher?.close(); + Reflect.set(NotificationsService, 'watcher', null); + } + + async function add(type: NotificationType, id: string) { + await Reflect.get(service, 'handleNotificationAdd').call(service, file(type, id)); + } + + async function expectCounts(unread: number, archive: number) { + const overview = service.getOverview(); + expect(overview.unread.total).toBe(unread); + expect(overview.archive.total).toBe(archive); + for (const type of Object.values(NotificationType)) { + const notifications = await service.getNotifications({ type, offset: 0, limit: 100 }); + expect(overview[type.toLowerCase()].total).toBe(notifications.length); + } + } + + beforeEach(async () => { + basePath = await mkdtemp(join(tmpdir(), 'notification-overview-')); + await Promise.all( + Object.values(NotificationType).map((type) => mkdir(join(basePath, type.toLowerCase()))) + ); + }); + + afterEach(async () => { + await stopWatcher(); + await Reflect.get(service, 'overviewRefresh'); + await rm(basePath, { recursive: true, force: true }); + }); + + it('hydrates counts with the same duplicate filtering as the lists', async () => { + for (const importance of ['normal', 'warning', 'alert']) { + await write(NotificationType.UNREAD, `${importance}.notify`, importance); + await write(NotificationType.ARCHIVE, `${importance}.notify`, importance); + await write(NotificationType.ARCHIVE, `archived-${importance}.notify`, importance); + } + await start(); + await expectCounts(3, 3); + expect(service.getOverview().archive).toEqual({ info: 1, warning: 1, alert: 1, total: 3 }); + await service.recalculateOverview(); + await expectCounts(3, 3); + }); + + it.each([NotificationType.UNREAD, NotificationType.ARCHIVE])( + 'handles legacy copies written %s first and repeated events', + async (first) => { + await start(); + const second = + first === NotificationType.UNREAD ? NotificationType.ARCHIVE : NotificationType.UNREAD; + await write(first, 'legacy.notify'); + await add(first, 'legacy.notify'); + await write(second, 'legacy.notify'); + await add(second, 'legacy.notify'); + await add(first, 'legacy.notify'); + await expectCounts(1, 0); + } + ); + + it('archives over an existing legacy archive copy and restores it', async () => { + await write(NotificationType.UNREAD, 'legacy.notify'); + await write(NotificationType.ARCHIVE, 'legacy.notify'); + await start(); + await service.archiveNotification({ id: 'legacy.notify' }); + await expectCounts(0, 1); + await service.markAsUnread({ id: 'legacy.notify' }); + await expectCounts(1, 0); + }); + + it('reveals the archive copy when unread is deleted', async () => { + await write(NotificationType.UNREAD, 'legacy.notify'); + await write(NotificationType.ARCHIVE, 'legacy.notify'); + await start(); + await service.deleteNotification({ id: 'legacy.notify', type: NotificationType.UNREAD }); + await expectCounts(0, 1); + }); + + it('does not decrement the archive count when a hidden copy is deleted', async () => { + await write(NotificationType.UNREAD, 'legacy.notify'); + await write(NotificationType.ARCHIVE, 'legacy.notify'); + await start(); + await service.deleteNotification({ id: 'legacy.notify', type: NotificationType.ARCHIVE }); + await expectCounts(1, 0); + }); + + it('recounts both categories after bulk deletes', async () => { + await write(NotificationType.UNREAD, 'legacy.notify'); + await write(NotificationType.ARCHIVE, 'legacy.notify'); + await start(); + await service.deleteNotifications(NotificationType.UNREAD); + await expectCounts(0, 1); + await service.deleteAllNotifications(); + await expectCounts(0, 0); + }); + + it('serializes snapshots and coalesces requests queued during a refresh', async () => { + await start(); + const snapshot = Reflect.get(service, 'buildOverviewSnapshot').bind(service); + const captured = Promise.withResolvers(); + const release = Promise.withResolvers(); + let calls = 0; + Reflect.set(service, 'buildOverviewSnapshot', async () => { + calls++; + const result = await snapshot(); + if (calls === 1) { + captured.resolve(); + await release.promise; + } + return result; + }); + const first = service.recalculateOverview(); + await captured.promise; + await write(NotificationType.UNREAD, 'during-refresh.notify'); + const queued = Array.from({ length: 20 }, () => service.recalculateOverview()); + release.resolve(); + await Promise.all([first, ...queued]); + expect(calls).toBe(2); + await expectCounts(1, 0); + }); + + it('can refresh again after a snapshot failure', async () => { + await start(); + const snapshot = Reflect.get(service, 'buildOverviewSnapshot'); + Reflect.set(service, 'buildOverviewSnapshot', () => Promise.reject(new Error('read failure'))); + await expect(service.recalculateOverview()).rejects.toThrow(); + Reflect.set(service, 'buildOverviewSnapshot', snapshot); + await write(NotificationType.UNREAD, 'after-failure.notify'); + await service.recalculateOverview(); + await expectCounts(1, 0); + }); + + it('handles new events while draining adds queued during hydration', async () => { + await start(); + Reflect.set(service, 'getNotificationsWatcher', async () => null); + const snapshot = Reflect.get(service, 'buildOverviewSnapshot').bind(service); + const processAdd = Reflect.get(service, 'processNotificationAdd').bind(service); + let firstSnapshot = true; + Reflect.set(service, 'buildOverviewSnapshot', async () => { + const result = await snapshot(); + if (firstSnapshot) { + firstSnapshot = false; + await write(NotificationType.UNREAD, 'queued.notify'); + await add(NotificationType.UNREAD, 'queued.notify'); + } + return result; + }); + Reflect.set(service, 'processNotificationAdd', async (path: string) => { + await processAdd(path); + if (path.endsWith('/queued.notify')) { + await write(NotificationType.UNREAD, 'late.notify'); + await add(NotificationType.UNREAD, 'late.notify'); + } + }); + await Reflect.get(service, 'initializeNotificationsState').call(service, basePath); + await expectCounts(2, 0); + }); + + it('tracks external writes and unlink events through the real watcher', async () => { + await start(true); + const watcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + await new Promise((resolve) => watcher.once('ready', resolve)); + await write(NotificationType.ARCHIVE, 'external.notify'); + await vi.waitFor(() => expect(service.getOverview().archive.total).toBe(1)); + await write(NotificationType.UNREAD, 'external.notify'); + await vi.waitFor(() => + expect(service.getOverview()).toMatchObject({ unread: { total: 1 }, archive: { total: 0 } }) + ); + await unlink(file(NotificationType.UNREAD, 'external.notify')); + await vi.waitFor(() => + expect(service.getOverview()).toMatchObject({ unread: { total: 0 }, archive: { total: 1 } }) + ); + await expectCounts(0, 1); + }); +}); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts index 04834ecef8..4325fcd7fa 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts @@ -43,6 +43,12 @@ export class NotificationsService { */ private path: string | null = null; private initialization: Promise | null = null; + private overviewRefresh: Promise = Promise.resolve(); + private pendingOverviewRefresh: Promise<{ + errorOccurred: boolean; + overview: NotificationOverview; + seenPaths: Set; + }> | null = null; private isHydratingOverview = false; private pendingNotificationAdds = new Set(); @@ -106,14 +112,13 @@ export class NotificationsService { try { await this.getNotificationsWatcher(basePath, recreate); - const { errorOccurred, overview, seenPaths } = await this.buildOverviewSnapshot(); - NotificationsService.overview = overview; - await this.publishOverview(); + const { errorOccurred, seenPaths } = await this.refreshOverview(); const pendingAdds = [...this.pendingNotificationAdds].filter( (path) => !seenPaths.has(path) ); this.pendingNotificationAdds.clear(); + this.isHydratingOverview = false; for (const path of pendingAdds) { await this.processNotificationAdd(path); @@ -162,9 +167,13 @@ export class NotificationsService { NotificationsService.watcher = watch(basePath, { usePolling: CHOKIDAR_USEPOLLING, ignoreInitial: true, - }).on('add', (path) => { - void this.handleNotificationAdd(path).catch((e) => this.logger.error(e)); - }); + }) + .on('add', (path) => { + void this.handleNotificationAdd(path).catch((e) => this.logger.error(e)); + }) + .on('unlink', () => { + void this.recalculateOverview().catch((e) => this.logger.error(e)); + }); return NotificationsService.watcher; } @@ -200,10 +209,9 @@ export class NotificationsService { throw error; } - this.increment(notification.importance, NotificationsService.overview[type.toLowerCase()]); + await this.recalculateOverview(); if (type === NotificationType.UNREAD) { - this.publishOverview(); pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_ADDED, { notificationAdded: notification, }); @@ -248,16 +256,29 @@ export class NotificationsService { collector['total'] += 1; } - private decrement(importance: NotificationImportance, collector: NotificationCounts) { - collector[importance.toLowerCase()] -= 1; - collector['total'] -= 1; + public async recalculateOverview() { + const { errorOccurred, overview } = await this.refreshOverview(); + return { error: errorOccurred, overview: structuredClone(overview) }; } - public async recalculateOverview() { - const { errorOccurred, overview } = await this.buildOverviewSnapshot(); - NotificationsService.overview = overview; - await this.publishOverview(); - return { error: errorOccurred, overview: this.getOverview() }; + private refreshOverview() { + if (this.pendingOverviewRefresh) { + return this.pendingOverviewRefresh; + } + + const refresh = this.overviewRefresh + .catch(() => undefined) + .then(async () => { + // Coalesce queued events, but refresh again for changes during an active snapshot. + this.pendingOverviewRefresh = null; + const snapshot = await this.buildOverviewSnapshot(); + NotificationsService.overview = snapshot.overview; + await this.publishOverview(); + return snapshot; + }); + this.overviewRefresh = refresh; + this.pendingOverviewRefresh = refresh; + return refresh; } private async buildOverviewSnapshot() { @@ -282,9 +303,9 @@ export class NotificationsService { // // recalculates stats for a particular notification type const recalculate = async (type: NotificationType) => { - const ids = await this.listFilesInFolder(this.paths()[type]); + const ids = await this.listNotificationFiles(type); ids.forEach((id) => seenPaths.add(id)); - const [notifications] = await this.loadNotificationsFromPaths(ids, {}); + const [notifications] = await this.loadNotificationsFromPaths(ids, { type }); notifications.forEach((n) => this.increment(n.importance, overview[type.toLowerCase()])); }; @@ -408,8 +429,7 @@ export class NotificationsService { const notification = await this.loadNotificationFile(path, type); await unlink(path); - this.decrement(notification.importance, NotificationsService.overview[type.toLowerCase()]); - await this.publishOverview(); + await this.recalculateOverview(); if (type === NotificationType.UNREAD) { void this.publishWarningsAndAlerts(); } @@ -427,13 +447,7 @@ export class NotificationsService { */ public async deleteNotifications(type: NotificationType) { await emptyDir(this.paths()[type]); - NotificationsService.overview[type.toLowerCase()] = { - alert: 0, - info: 0, - warning: 0, - total: 0, - }; - await this.publishOverview(); + await this.recalculateOverview(); if (type === NotificationType.UNREAD) { void this.publishWarningsAndAlerts(); } @@ -456,68 +470,14 @@ export class NotificationsService { * CRUD: Updating Notifications *------------------------------------------------------------------------**/ - /** - * Returns a function that: - * 1. moves a notification from one category to another. - * 2. updates stats overview - * 3. updates the stats snapshot if provided - * - * Note: the returned function implicitly triggers a pubsub event via `fs.rename`, - * which is expected to trigger `NOTIFICATION_ADDED` & `NOTIFICATION_OVERVIEW`. - * - * The published overview will include the update from this operation. - * - * @param params - * @returns lambda function - */ - private moveNotification(params: { - from: NotificationType; - to: NotificationType; - snapshot?: NotificationOverview; - }) { - const { from, to, snapshot } = params; - const fromStatKey = from.toLowerCase(); - const toStatKey = to.toLowerCase(); + private moveNotification(params: { from: NotificationType; to: NotificationType }) { + const { from, to } = params; return async (notification: Notification) => { const currentPath = this.notificationPath(notification.id, from); const targetPath = this.notificationPath(notification.id, to); - - /**----------------------- - * Event, PubSub, & Overview Update logic - * - * We assume `rename` kicks off 'unlink' and 'add' events - * in the chokidar file watcher (see `getNotificationsWatcher`). - * - * We assume the 'add' handler publishes to - * NOTIFICATION_ADDED & NOTIFICATION_OVERVIEW, - * and that no pubsub or overview updates occur upon 'unlink'. - * - * Thus, we explicitly update our state here via `decrement` and implicitly expect - * it to be updated (i.e. incremented & pubsub'd) via our filesystem changes. - * - * The reasons for this discrepancy are: - * - Backwards compatibility: not every notification will be created through this API, - * so we track state by watching the store (i.e. the file system). - * - * - Technical Limitations: By the time the unlink event fires, the notification file - * can no longer be read. This means we can only track overview totals accurately; - * to track other stats, we have to update them manually, prior to file deletion. - *------------------------**/ - this.decrement(notification.importance, NotificationsService.overview[fromStatKey]); - try { - await rename(currentPath, targetPath); - } catch (err) { - // revert our earlier decrement - // we do it this way (decrement -> try rename -> revert if error) to avoid - // a race condition between `rename` and `decrement` - this.increment(notification.importance, NotificationsService.overview[fromStatKey]); - throw err; - } - - if (snapshot) { - this.decrement(notification.importance, snapshot[fromStatKey]); - this.increment(notification.importance, snapshot[toStatKey]); - } + await rename(currentPath, targetPath); + // A legacy archive copy may already exist, so rename need not emit an add event. + await this.recalculateOverview(); }; } @@ -531,21 +491,10 @@ export class NotificationsService { throw new AppError(`Could not find notification in unreads: ${id}`, 404); } - /**----------------------- - * Why we use a snapshot - * - * An implicit update to `overview` creates a race condition: - * it might be missing changes from the 'add' event (i.e. incrementing the notification's new category). - * - * So, we use & modify a snapshot of the overview to make sure we're returning accurate - * data to the client. - *------------------------**/ - const snapshot = this.getOverview(); const notification = await this.loadNotificationFile(unreadPath, NotificationType.UNREAD); const moveToArchive = this.moveNotification({ from: NotificationType.UNREAD, to: NotificationType.ARCHIVE, - snapshot, }); await moveToArchive(notification); @@ -565,14 +514,10 @@ export class NotificationsService { throw new AppError(`Could not find notification in archive: ${id}`, 404); } - // we use a snapshot to provide an accurate overview update - // otherwise, we'd enter a race condition with the 'add' file watcher event handler - const snapshot = this.getOverview(); const notification = await this.loadNotificationFile(archivePath, NotificationType.ARCHIVE); const moveToUnread = this.moveNotification({ from: NotificationType.ARCHIVE, to: NotificationType.UNREAD, - snapshot, }); await moveToUnread(notification); @@ -591,18 +536,16 @@ export class NotificationsService { return { overview: NotificationsService.overview }; } - const overviewSnapshot = this.getOverview(); const unreads = await this.listFilesInFolder(UNREAD); const [notifications] = await this.loadNotificationsFromPaths(unreads, { importance }); const archive = this.moveNotification({ from: NotificationType.UNREAD, to: NotificationType.ARCHIVE, - snapshot: overviewSnapshot, }); const stats = await batchProcess(notifications, archive); void this.publishWarningsAndAlerts(); - return { ...stats, overview: overviewSnapshot }; + return { ...stats, overview: this.getOverview() }; } public async unarchiveAll(importance?: NotificationImportance) { @@ -614,18 +557,16 @@ export class NotificationsService { return { overview: NotificationsService.overview }; } - const overviewSnapshot = this.getOverview(); const archives = await this.listFilesInFolder(ARCHIVE); const [notifications] = await this.loadNotificationsFromPaths(archives, { importance }); const unArchive = this.moveNotification({ from: NotificationType.ARCHIVE, to: NotificationType.UNREAD, - snapshot: overviewSnapshot, }); const stats = await batchProcess(notifications, unArchive); void this.publishWarningsAndAlerts(); - return { ...stats, overview: overviewSnapshot }; + return { ...stats, overview: this.getOverview() }; } /** @@ -669,23 +610,23 @@ export class NotificationsService { this.logger.verbose('Getting Notifications'); const { type = NotificationType.UNREAD } = filters; - const { ARCHIVE, UNREAD } = this.paths(); - let files: string[]; + const files = await this.listNotificationFiles(type); + const [notifications] = await this.loadNotificationsFromPaths(files, filters); + return notifications; + } + + private async listNotificationFiles(type: NotificationType): Promise { + const { ARCHIVE, UNREAD } = this.paths(); if (type === NotificationType.UNREAD) { - files = await this.listFilesInFolder(UNREAD); - } else { - // Exclude notifications present in both unread & archive from archive. - //* this is necessary because the legacy script writes new notifications to both places. - //* this should be a temporary measure. - const unreads = new Set(await readdir(UNREAD)); - files = await this.listFilesInFolder(ARCHIVE, (archives) => { - return archives.filter((file) => !unreads.has(file)); - }); + return this.listFilesInFolder(UNREAD); } - const [notifications] = await this.loadNotificationsFromPaths(files, filters); - return notifications; + // The legacy notifier writes each new notification to both directories. + const unreads = new Set(await readdir(UNREAD)); + return this.listFilesInFolder(ARCHIVE, (archives) => + archives.filter((file) => !unreads.has(file)) + ); } /** From a368d11719f76c55973f1df2dc74875414841799 Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sun, 25 Oct 2026 06:28:37 -0400 Subject: [PATCH 3/8] fix(notifications): reconcile watcher events incrementally --- .../notifications.overview.spec.ts | 597 ++++++++++++++---- .../notifications.service.spec.ts | 3 +- .../notifications/notifications.service.ts | 236 ++++--- 3 files changed, 629 insertions(+), 207 deletions(-) diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts index 2ea664c78c..9648611ffd 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts @@ -1,17 +1,38 @@ +import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo'; import { ConfigService } from '@nestjs/config'; -import { mkdir, mkdtemp, rm, unlink, writeFile } from 'fs/promises'; +import { GraphQLModule } from '@nestjs/graphql'; +import { FastifyAdapter, NestFastifyApplication } from '@nestjs/platform-fastify'; +import { Test } from '@nestjs/testing'; +import { copyFile, mkdir, mkdtemp, readFile, rename, rm, unlink, writeFile } from 'fs/promises'; import { tmpdir } from 'os'; import { join } from 'path'; import type { FSWatcher } from 'chokidar'; +import { PrefixedID } from '@unraid/shared/prefixed-id-scalar.js'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { NotificationType } from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { pubsub, PUBSUB_CHANNEL } from '@app/core/pubsub.js'; +import { + Notification, + NotificationImportance, + NotificationOverview, + NotificationType, +} from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsResolver } from '@app/unraid-api/graph/resolvers/notifications/notifications.resolver.js'; import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; -describe.sequential('Notification overview visibility', () => { +const { UNREAD, ARCHIVE } = NotificationType; + +describe.sequential('Notification filesystem and subscription integration', () => { let basePath: string; let service: NotificationsService; + let config: ConfigService; + let watcher: FSWatcher; + let subscriptions: number[]; + let overviews: NotificationOverview[]; + let additions: Notification[]; + let warnings: Notification[][]; + let events: { event: string; path: string }[]; const file = (type: NotificationType, id: string) => join(basePath, type.toLowerCase(), id); async function write(type: NotificationType, id: string, importance = 'normal') { @@ -21,187 +42,533 @@ describe.sequential('Notification overview visibility', () => { ); } - async function start(keepWatcher = false) { - service = new NotificationsService( - new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }) + async function start() { + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_OVERVIEW, + (payload: { notificationsOverview: NotificationOverview }) => + overviews.push(payload.notificationsOverview) + ) + ); + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_ADDED, + (payload: { notificationAdded: Notification }) => + additions.push(payload.notificationAdded) + ) + ); + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, + (payload: { notificationsWarningsAndAlerts: Notification[] }) => + warnings.push(payload.notificationsWarningsAndAlerts) + ) ); + service = new NotificationsService(config); await Reflect.get(service, 'initialization'); - if (!keepWatcher) await stopWatcher(); - } - - async function stopWatcher() { - const watcher: FSWatcher | null = Reflect.get(NotificationsService, 'watcher'); - await watcher?.close(); - Reflect.set(NotificationsService, 'watcher', null); + watcher = Reflect.get(NotificationsService, 'watcher'); + watcher.on('all', (event, path) => events.push({ event, path })); } - async function add(type: NotificationType, id: string) { - await Reflect.get(service, 'handleNotificationAdd').call(service, file(type, id)); + async function expectCounts(unread: number, archive: number) { + await vi.waitFor( + async () => { + const overview = service.getOverview(); + expect(overview.unread.total).toBe(unread); + expect(overview.archive.total).toBe(archive); + for (const type of Object.values(NotificationType)) { + const notifications = await service.getNotifications({ + type, + offset: 0, + limit: 1000, + }); + const counts = overview[type.toLowerCase()]; + expect(counts.total).toBe(notifications.length); + for (const importance of Object.values(NotificationImportance)) { + expect(counts[importance.toLowerCase()]).toBe( + notifications.filter( + (notification) => notification.importance === importance + ).length + ); + } + } + }, + { timeout: 4000, interval: 20 } + ); } - async function expectCounts(unread: number, archive: number) { - const overview = service.getOverview(); - expect(overview.unread.total).toBe(unread); - expect(overview.archive.total).toBe(archive); - for (const type of Object.values(NotificationType)) { - const notifications = await service.getNotifications({ type, offset: 0, limit: 100 }); - expect(overview[type.toLowerCase()].total).toBe(notifications.length); - } + async function expectEvent(event: string, type: NotificationType, id: string) { + await vi.waitFor(() => expect(events).toContainEqual({ event, path: file(type, id) }), { + timeout: 4000, + }); + events.splice( + events.findIndex((entry) => entry.event === event && entry.path === file(type, id)), + 1 + ); + await Reflect.get(service, 'overviewUpdate'); } beforeEach(async () => { - basePath = await mkdtemp(join(tmpdir(), 'notification-overview-')); + basePath = await mkdtemp(join(tmpdir(), 'notification-integration-')); await Promise.all( Object.values(NotificationType).map((type) => mkdir(join(basePath, type.toLowerCase()))) ); + config = new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }); + subscriptions = []; + overviews = []; + additions = []; + warnings = []; + events = []; }); afterEach(async () => { - await stopWatcher(); - await Reflect.get(service, 'overviewRefresh'); + const activeWatcher: FSWatcher | null = Reflect.get(NotificationsService, 'watcher'); + await activeWatcher?.close(); + Reflect.set(NotificationsService, 'watcher', null); + if (service) await Reflect.get(service, 'overviewUpdate'); + subscriptions.forEach((subscription) => pubsub.unsubscribe(subscription)); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); await rm(basePath, { recursive: true, force: true }); }); - it('hydrates counts with the same duplicate filtering as the lists', async () => { + it('hydrates all severities and excludes legacy archive copies', async () => { for (const importance of ['normal', 'warning', 'alert']) { - await write(NotificationType.UNREAD, `${importance}.notify`, importance); - await write(NotificationType.ARCHIVE, `${importance}.notify`, importance); - await write(NotificationType.ARCHIVE, `archived-${importance}.notify`, importance); + await write(UNREAD, `${importance}.notify`, importance); + await write(ARCHIVE, `${importance}.notify`, importance); + await write(ARCHIVE, `archived-${importance}.notify`, importance); } await start(); await expectCounts(3, 3); expect(service.getOverview().archive).toEqual({ info: 1, warning: 1, alert: 1, total: 3 }); + expect(additions).toEqual([]); await service.recalculateOverview(); await expectCounts(3, 3); }); - it.each([NotificationType.UNREAD, NotificationType.ARCHIVE])( - 'handles legacy copies written %s first and repeated events', + it.each([UNREAD, ARCHIVE])( + 'handles legacy files written %s first without recounting', async (first) => { await start(); - const second = - first === NotificationType.UNREAD ? NotificationType.ARCHIVE : NotificationType.UNREAD; - await write(first, 'legacy.notify'); - await add(first, 'legacy.notify'); - await write(second, 'legacy.notify'); - await add(second, 'legacy.notify'); - await add(first, 'legacy.notify'); + const recount = vi.spyOn(service, 'recalculateOverview'); + const second = first === UNREAD ? ARCHIVE : UNREAD; + const initialOverview = structuredClone(overviews[0]); + await write(first, 'legacy.notify', 'warning'); + await expectEvent('add', first, 'legacy.notify'); + await expectCounts(first === UNREAD ? 1 : 0, first === ARCHIVE ? 1 : 0); + await copyFile(file(first, 'legacy.notify'), file(second, 'legacy.notify')); + await expectEvent('add', second, 'legacy.notify'); await expectCounts(1, 0); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + await vi.waitFor(() => expect(warnings.at(-1)).toHaveLength(1)); + expect(recount).not.toHaveBeenCalled(); + expect(overviews.at(-1)).toEqual(service.getOverview()); + expect(overviews[0]).toEqual(initialOverview); } ); - it('archives over an existing legacy archive copy and restores it', async () => { - await write(NotificationType.UNREAD, 'legacy.notify'); - await write(NotificationType.ARCHIVE, 'legacy.notify'); + it('creates a notification through the public API and publishes its real file data', async () => { await start(); - await service.archiveNotification({ id: 'legacy.notify' }); - await expectCounts(0, 1); - await service.markAsUnread({ id: 'legacy.notify' }); + const notification = await service.createNotification({ + title: 'Integration notification', + subject: 'Disk status', + description: 'Test warning', + importance: NotificationImportance.WARNING, + }); + expect(await readFile(file(UNREAD, notification.id), 'utf8')).toContain('importance=warning'); + await expectEvent('add', UNREAD, notification.id); await expectCounts(1, 0); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + expect(additions[0]).toMatchObject({ + id: notification.id, + subject: 'Disk status', + importance: NotificationImportance.WARNING, + }); }); - it('reveals the archive copy when unread is deleted', async () => { - await write(NotificationType.UNREAD, 'legacy.notify'); - await write(NotificationType.ARCHIVE, 'legacy.notify'); + it('archives over a legacy copy, restores, and deletes without late watcher drift', async () => { + await write(UNREAD, 'legacy.notify', 'alert'); + await write(ARCHIVE, 'legacy.notify', 'normal'); await start(); - await service.deleteNotification({ id: 'legacy.notify', type: NotificationType.UNREAD }); + const recount = vi.spyOn(service, 'recalculateOverview'); + const archived = await service.archiveNotification({ id: 'legacy.notify' }); + expect(archived.type).toBe(ARCHIVE); + expect(service.getOverview().archive.alert).toBe(1); + await expectEvent('unlink', UNREAD, 'legacy.notify'); await expectCounts(0, 1); + await vi.waitFor(() => expect(warnings.at(-1)).toEqual([])); + const restored = await service.markAsUnread({ id: 'legacy.notify' }); + expect(restored.type).toBe(UNREAD); + expect(service.getOverview().unread.alert).toBe(1); + await expectEvent('unlink', ARCHIVE, 'legacy.notify'); + await expectEvent('add', UNREAD, 'legacy.notify'); + await expectCounts(1, 0); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + expect(additions[0]).toMatchObject({ + id: 'legacy.notify', + type: UNREAD, + importance: NotificationImportance.ALERT, + }); + await service.deleteNotification({ id: 'legacy.notify', type: UNREAD }); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 0); + await vi.waitFor(() => expect(warnings.at(-1)).toEqual([])); + expect(recount).not.toHaveBeenCalled(); }); - it('does not decrement the archive count when a hidden copy is deleted', async () => { - await write(NotificationType.UNREAD, 'legacy.notify'); - await write(NotificationType.ARCHIVE, 'legacy.notify'); + it('updates warning subscription content without changing counts or adding again', async () => { await start(); - await service.deleteNotification({ id: 'legacy.notify', type: NotificationType.ARCHIVE }); + await write(UNREAD, 'content.notify', 'warning'); + await expectEvent('add', UNREAD, 'content.notify'); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + await writeFile( + file(UNREAD, 'content.notify'), + 'timestamp=1700000000\nevent=Test\nsubject=Updated subject\ndescription=Updated description\nimportance=warning\n' + ); + await expectEvent('change', UNREAD, 'content.notify'); + await vi.waitFor(() => + expect(warnings.at(-1)?.[0]).toMatchObject({ + subject: 'Updated subject', + description: 'Updated description', + }) + ); await expectCounts(1, 0); + expect(additions).toHaveLength(1); }); - it('recounts both categories after bulk deletes', async () => { - await write(NotificationType.UNREAD, 'legacy.notify'); - await write(NotificationType.ARCHIVE, 'legacy.notify'); + it('tracks external overwrite renames and severity changes', async () => { + await write(UNREAD, 'external.notify', 'warning'); + await write(ARCHIVE, 'external.notify', 'normal'); await start(); - await service.deleteNotifications(NotificationType.UNREAD); + await rename(file(UNREAD, 'external.notify'), file(ARCHIVE, 'external.notify')); + await expectEvent('unlink', UNREAD, 'external.notify'); + await expectCounts(0, 1); + expect(service.getOverview().archive.warning).toBe(1); + await write(ARCHIVE, 'external.notify', 'alert'); + await expectEvent('change', ARCHIVE, 'external.notify'); + await vi.waitFor(() => expect(service.getOverview().archive.alert).toBe(1)); await expectCounts(0, 1); + await rename(file(ARCHIVE, 'external.notify'), file(UNREAD, 'external.notify')); + await expectEvent('add', UNREAD, 'external.notify'); + await expectCounts(1, 0); + }); + + it.each([UNREAD, ARCHIVE])('deletes a %s copy according to actual visibility', async (type) => { + await write(UNREAD, 'legacy.notify', 'warning'); + await write(ARCHIVE, 'legacy.notify', 'alert'); + await start(); + await service.deleteNotification({ id: 'legacy.notify', type }); + await expectEvent('unlink', type, 'legacy.notify'); + await expectCounts(type === UNREAD ? 0 : 1, type === UNREAD ? 1 : 0); + if (type === UNREAD) expect(service.getOverview().archive.alert).toBe(1); + }); + + it('handles external deletion of either legacy copy', async () => { + await write(UNREAD, 'legacy.notify', 'warning'); + await write(ARCHIVE, 'legacy.notify', 'alert'); + await start(); + await unlink(file(UNREAD, 'legacy.notify')); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 1); + expect(service.getOverview().archive.alert).toBe(1); + await unlink(file(ARCHIVE, 'legacy.notify')); + await expectEvent('unlink', ARCHIVE, 'legacy.notify'); + await expectCounts(0, 0); + }); + + it('archives/restores by importance, archives all, then clears both categories', async () => { + for (const importance of ['normal', 'warning', 'alert']) { + await write(UNREAD, `${importance}.notify`, importance); + await write(ARCHIVE, `${importance}.notify`, importance); + } + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + await service.archiveAll(NotificationImportance.WARNING); + await expectCounts(2, 1); + expect(service.getOverview().archive.warning).toBe(1); + await service.unarchiveAll(NotificationImportance.WARNING); + await expectCounts(3, 0); + await service.archiveAll(); + await expectCounts(0, 3); + await service.unarchiveAll(); + await expectCounts(3, 0); await service.deleteAllNotifications(); await expectCounts(0, 0); + expect(recount).not.toHaveBeenCalled(); }); - it('serializes snapshots and coalesces requests queued during a refresh', async () => { + it('reveals hidden archive files after a bulk unread delete', async () => { + await write(UNREAD, 'legacy.notify', 'warning'); + await write(ARCHIVE, 'legacy.notify', 'alert'); await start(); - const snapshot = Reflect.get(service, 'buildOverviewSnapshot').bind(service); - const captured = Promise.withResolvers(); - const release = Promise.withResolvers(); - let calls = 0; - Reflect.set(service, 'buildOverviewSnapshot', async () => { - calls++; - const result = await snapshot(); - if (calls === 1) { - captured.resolve(); - await release.promise; - } - return result; + await service.deleteNotifications(UNREAD); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 1); + expect(service.getOverview().archive.alert).toBe(1); + }); + + it('reconciles cached IDs even when their unlink event has not arrived', async () => { + await write(UNREAD, 'removed.notify'); + await start(); + await watcher.close(); + await unlink(file(UNREAD, 'removed.notify')); + await service.deleteNotifications(UNREAD); + await expectCounts(0, 0); + }); + + it('handles a burst of legacy writes without full recounts or duplicate additions', async () => { + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + const ids = Array.from({ length: 24 }, (_, index) => `burst-${index}.notify`); + await Promise.all( + ids.map(async (id) => { + await write(UNREAD, id, 'warning'); + await copyFile(file(UNREAD, id), file(ARCHIVE, id)); + }) + ); + await expectCounts(24, 0); + await vi.waitFor(() => expect(additions).toHaveLength(24)); + await vi.waitFor(() => expect(events.filter((entry) => entry.event === 'add')).toHaveLength(48)); + await Reflect.get(service, 'overviewUpdate'); + await expectCounts(24, 0); + expect(new Set(additions.map((notification) => notification.id)).size).toBe(24); + await vi.waitFor(() => expect(warnings.at(-1)).toHaveLength(1)); + expect(recount).not.toHaveBeenCalled(); + }); + + it('bounds event reads to affected files even with a large archived history', async () => { + await Promise.all( + Array.from({ length: 128 }, (_, index) => write(ARCHIVE, `history-${index}.notify`)) + ); + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + const reads: string[] = []; + const load = Reflect.get(service, 'loadNotificationFile').bind(service); + Reflect.set(service, 'loadNotificationFile', (path: string, type: NotificationType) => { + reads.push(path); + return load(path, type); }); - const first = service.recalculateOverview(); - await captured.promise; - await write(NotificationType.UNREAD, 'during-refresh.notify'); - const queued = Array.from({ length: 20 }, () => service.recalculateOverview()); - release.resolve(); - await Promise.all([first, ...queued]); - expect(calls).toBe(2); - await expectCounts(1, 0); + await write(UNREAD, 'new.notify'); + await copyFile(file(UNREAD, 'new.notify'), file(ARCHIVE, 'new.notify')); + await expectEvent('add', UNREAD, 'new.notify'); + await expectEvent('add', ARCHIVE, 'new.notify'); + await vi.waitFor(() => + expect(service.getOverview()).toMatchObject({ + unread: { total: 1 }, + archive: { total: 128 }, + }) + ); + expect(recount).not.toHaveBeenCalled(); + expect(reads.some((path) => path.includes('/history-'))).toBe(false); + await expectCounts(1, 128); }); - it('can refresh again after a snapshot failure', async () => { + it('retains the previous counts when an explicit reconciliation cannot read a category', async () => { + await write(UNREAD, 'visible.notify'); await start(); - const snapshot = Reflect.get(service, 'buildOverviewSnapshot'); - Reflect.set(service, 'buildOverviewSnapshot', () => Promise.reject(new Error('read failure'))); - await expect(service.recalculateOverview()).rejects.toThrow(); - Reflect.set(service, 'buildOverviewSnapshot', snapshot); - await write(NotificationType.UNREAD, 'after-failure.notify'); + await watcher.close(); + const unread = file(UNREAD, ''); + const saved = join(basePath, 'saved-unread'); + await rename(unread, saved); + try { + const before = service.getOverview(); + const result = await service.recalculateOverview(); + expect(result.error).toBe(true); + expect(service.getOverview()).toEqual(before); + } finally { + await rename(saved, unread); + } await service.recalculateOverview(); await expectCounts(1, 0); }); - it('handles new events while draining adds queued during hydration', async () => { + it('keeps counts intact after a failed rename and allows a successful retry', async () => { + await write(UNREAD, 'blocked.notify', 'warning'); await start(); - Reflect.set(service, 'getNotificationsWatcher', async () => null); - const snapshot = Reflect.get(service, 'buildOverviewSnapshot').bind(service); - const processAdd = Reflect.get(service, 'processNotificationAdd').bind(service); - let firstSnapshot = true; + await mkdir(file(ARCHIVE, 'blocked.notify')); + await expect(service.archiveNotification({ id: 'blocked.notify' })).rejects.toThrow(); + await expectCounts(1, 0); + await rm(file(ARCHIVE, 'blocked.notify'), { recursive: true }); + await service.archiveNotification({ id: 'blocked.notify' }); + await expectEvent('unlink', UNREAD, 'blocked.notify'); + await expectCounts(0, 1); + }); + + it('replays real watcher changes after an in-flight full reconciliation', async () => { + await write(UNREAD, 'removed.notify'); + await start(); + const original = Reflect.get(service, 'buildOverviewSnapshot').bind(service); + const captured = Promise.withResolvers(); + const release = Promise.withResolvers(); Reflect.set(service, 'buildOverviewSnapshot', async () => { - const result = await snapshot(); - if (firstSnapshot) { - firstSnapshot = false; - await write(NotificationType.UNREAD, 'queued.notify'); - await add(NotificationType.UNREAD, 'queued.notify'); - } + const result = await original(); + captured.resolve(); + await release.promise; return result; }); - Reflect.set(service, 'processNotificationAdd', async (path: string) => { - await processAdd(path); - if (path.endsWith('/queued.notify')) { - await write(NotificationType.UNREAD, 'late.notify'); - await add(NotificationType.UNREAD, 'late.notify'); - } + const reconciliation = service.recalculateOverview(); + try { + await captured.promise; + await write(UNREAD, 'added.notify', 'alert'); + await unlink(file(UNREAD, 'removed.notify')); + await vi.waitFor( + () => { + expect(events).toContainEqual({ event: 'add', path: file(UNREAD, 'added.notify') }); + expect(events).toContainEqual({ + event: 'unlink', + path: file(UNREAD, 'removed.notify'), + }); + }, + { timeout: 4000 } + ); + } finally { + release.resolve(); + await reconciliation; + } + await Reflect.get(service, 'overviewUpdate'); + await expectCounts(1, 0); + expect(service.getOverview().unread.alert).toBe(1); + expect(overviews.at(-1)).toEqual(service.getOverview()); + }); + + it('replays actual adds and removals that arrive during the startup snapshot', async () => { + await write(UNREAD, 'old.notify'); + await write(ARCHIVE, 'old.notify', 'warning'); + const prototype = NotificationsService.prototype; + const snapshot = Reflect.get(prototype, 'buildOverviewSnapshot'); + const release = Promise.withResolvers(); + let captured = false; + Reflect.set(prototype, 'buildOverviewSnapshot', async function (this: NotificationsService) { + const result = await snapshot.call(this); + captured = true; + await release.promise; + return result; }); - await Reflect.get(service, 'initializeNotificationsState').call(service, basePath); - await expectCounts(2, 0); + const starting = start(); + try { + await vi.waitFor(() => expect(captured).toBe(true)); + const activeWatcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + const duringStartup: string[] = []; + activeWatcher.on('all', (event, path) => duringStartup.push(`${event}:${path}`)); + await write(UNREAD, 'during-startup.notify', 'alert'); + await unlink(file(UNREAD, 'old.notify')); + await vi.waitFor(() => { + expect(duringStartup).toContain(`add:${file(UNREAD, 'during-startup.notify')}`); + expect(duringStartup).toContain(`unlink:${file(UNREAD, 'old.notify')}`); + }); + await writeFile( + file(UNREAD, 'during-startup.notify'), + 'timestamp=1700000000\nevent=Test\nsubject=Startup update\ndescription=Test\nimportance=alert\n' + ); + await vi.waitFor(() => + expect(duringStartup).toContain(`change:${file(UNREAD, 'during-startup.notify')}`) + ); + } finally { + release.resolve(); + await starting; + Reflect.set(prototype, 'buildOverviewSnapshot', snapshot); + } + await expectCounts(1, 1); + expect(service.getOverview().unread.alert).toBe(1); + expect(service.getOverview().archive.warning).toBe(1); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + expect(additions[0]).toMatchObject({ id: 'during-startup.notify', subject: 'Startup update' }); }); - it('tracks external writes and unlink events through the real watcher', async () => { - await start(true); - const watcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); - await new Promise((resolve) => watcher.once('ready', resolve)); - await write(NotificationType.ARCHIVE, 'external.notify'); - await vi.waitFor(() => expect(service.getOverview().archive.total).toBe(1)); - await write(NotificationType.UNREAD, 'external.notify'); - await vi.waitFor(() => - expect(service.getOverview()).toMatchObject({ unread: { total: 1 }, archive: { total: 0 } }) + it('rebuilds the index and watcher when the configured directory changes', async () => { + await write(UNREAD, 'old.notify'); + await start(); + const previousWatcher = watcher; + const next = join(basePath, 'alternate'); + await Promise.all( + Object.values(NotificationType).map((type) => + mkdir(join(next, type.toLowerCase()), { recursive: true }) + ) ); - await unlink(file(NotificationType.UNREAD, 'external.notify')); - await vi.waitFor(() => - expect(service.getOverview()).toMatchObject({ unread: { total: 0 }, archive: { total: 1 } }) + const nextFile = join(next, 'archive', 'new.notify'); + await writeFile( + nextFile, + 'timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=alert\n' ); + config.set('store.dynamix.notify.path', next); + await service.getNotifications({ type: ARCHIVE, offset: 0, limit: 100 }); + expect(Object.keys(previousWatcher.getWatched())).toHaveLength(0); await expectCounts(0, 1); + expect(service.getOverview().archive.alert).toBe(1); + watcher = Reflect.get(NotificationsService, 'watcher'); + const removed = new Promise((resolve) => watcher.once('unlink', () => resolve())); + await unlink(nextFile); + await removed; + await expectCounts(0, 0); + }); + + it('returns matching lists and counts through actual GraphQL queries and mutations', async () => { + await write(UNREAD, 'graphql.notify', 'warning'); + await write(ARCHIVE, 'graphql.notify', 'warning'); + await start(); + vi.stubGlobal('getServerIdentifier', () => 'integration'); + const module = await Test.createTestingModule({ + imports: [ + GraphQLModule.forRoot({ + driver: ApolloDriver, + autoSchemaFile: true, + }), + ], + providers: [ + PrefixedID, + NotificationsResolver, + { provide: NotificationsService, useValue: service }, + ], + }).compile(); + const app = module.createNestApplication(new FastifyAdapter()); + try { + await app.init(); + await app.getHttpAdapter().getInstance().ready(); + const execute = async (query: string) => { + const response = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query }, + }); + expect(response.statusCode).toBe(200); + const result = response.json<{ data: unknown; errors?: unknown[] }>(); + expect(result.errors).toBeUndefined(); + return result.data; + }; + expect( + await execute( + '{ notifications { overview { unread { total } archive { total } } list(filter: {type: ARCHIVE, offset: 0, limit: 100}) { id } } }' + ) + ).toEqual({ + notifications: { overview: { unread: { total: 1 }, archive: { total: 0 } }, list: [] }, + }); + expect( + await execute( + 'mutation { archiveNotification(id: "integration:graphql.notify") { id type } }' + ) + ).toEqual({ archiveNotification: { id: 'integration:graphql.notify', type: ARCHIVE } }); + await expectEvent('unlink', UNREAD, 'graphql.notify'); + expect( + await execute( + '{ notifications { overview { unread { total } archive { total } } list(filter: {type: ARCHIVE, offset: 0, limit: 100}) { id importance } } }' + ) + ).toEqual({ + notifications: { + overview: { unread: { total: 0 }, archive: { total: 1 } }, + list: [ + { id: 'integration:graphql.notify', importance: NotificationImportance.WARNING }, + ], + }, + }); + expect( + await execute( + 'mutation { unreadNotification(id: "integration:graphql.notify") { type } }' + ) + ).toEqual({ unreadNotification: { type: UNREAD } }); + await expectEvent('add', UNREAD, 'graphql.notify'); + await expectCounts(1, 0); + } finally { + await app.close(); + } }); }); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts index 14448be631..8daeaadb67 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts @@ -58,7 +58,6 @@ async function disableNotificationsWatcher(service: NotificationsService) { } | null; await watcher?.close?.(); Reflect.set(NotificationsService, 'watcher', null); - Reflect.set(NotificationsService, 'overview', zeroOverview()); } } @@ -100,8 +99,8 @@ describe.sequential('NotificationsService', () => { // make sure each test is isolated (as much as possible) afterEach(async () => { - Reflect.set(NotificationsService, 'overview', zeroOverview()); await service.deleteAllNotifications(); + expect(service.getOverview()).toEqual(zeroOverview()); }); /**------------------------------------------------------------------------ diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts index 4325fcd7fa..c580f7a63c 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts @@ -1,7 +1,7 @@ import { Injectable, Logger } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { mkdir, readdir, readFile, rename, stat, unlink, writeFile } from 'fs/promises'; -import { basename, join } from 'path'; +import { basename, dirname, join } from 'path'; import type { Stats } from 'fs'; import { FSWatcher, watch } from 'chokidar'; @@ -33,6 +33,9 @@ import { validateObject } from '@app/unraid-api/graph/resolvers/validation.utils import { SortFn } from '@app/unraid-api/types/util.js'; import { batchProcess, formatDatetime, isFulfilled, isRejected, unraidTimestamp } from '@app/utils.js'; +type NotificationFileState = Partial>; +type NotificationFileEvent = 'add' | 'change' | 'unlink'; + @Injectable() export class NotificationsService { private static readonly FILE_IO_BATCH_SIZE = 32; @@ -43,14 +46,10 @@ export class NotificationsService { */ private path: string | null = null; private initialization: Promise | null = null; - private overviewRefresh: Promise = Promise.resolve(); - private pendingOverviewRefresh: Promise<{ - errorOccurred: boolean; - overview: NotificationOverview; - seenPaths: Set; - }> | null = null; + private overviewUpdate: Promise = Promise.resolve(); + private notificationFiles = new Map(); private isHydratingOverview = false; - private pendingNotificationAdds = new Set(); + private pendingNotificationEvents = new Map(); private static overview: NotificationOverview = { unread: { @@ -100,6 +99,11 @@ export class NotificationsService { }; } + private async waitForInitialization() { + this.paths(); + await this.initialization; + } + private notificationPath(id: string, type: NotificationType): string { return resolveFileInDirectory(this.paths()[type], id); } @@ -107,21 +111,18 @@ export class NotificationsService { private initializeNotificationsState(basePath: string, recreate = false) { const initialize = async () => { await this.ensureNotificationDirectories(basePath); - this.pendingNotificationAdds.clear(); + this.pendingNotificationEvents.clear(); this.isHydratingOverview = true; try { await this.getNotificationsWatcher(basePath, recreate); - const { errorOccurred, seenPaths } = await this.refreshOverview(); - - const pendingAdds = [...this.pendingNotificationAdds].filter( - (path) => !seenPaths.has(path) - ); - this.pendingNotificationAdds.clear(); + const { error: errorOccurred } = await this.reconcileOverview(); + const pendingEvents = [...this.pendingNotificationEvents]; + this.pendingNotificationEvents.clear(); this.isHydratingOverview = false; - for (const path of pendingAdds) { - await this.processNotificationAdd(path); + for (const [path, event] of pendingEvents) { + await this.handleNotificationEvent(path, event); } if (errorOccurred) { @@ -131,7 +132,7 @@ export class NotificationsService { } } finally { this.isHydratingOverview = false; - this.pendingNotificationAdds.clear(); + this.pendingNotificationEvents.clear(); } }; @@ -164,59 +165,107 @@ export class NotificationsService { } await NotificationsService.watcher?.close().catch((e) => this.logger.error(e)); - NotificationsService.watcher = watch(basePath, { + const watcher = watch(basePath, { usePolling: CHOKIDAR_USEPOLLING, ignoreInitial: true, - }) - .on('add', (path) => { - void this.handleNotificationAdd(path).catch((e) => this.logger.error(e)); - }) - .on('unlink', () => { - void this.recalculateOverview().catch((e) => this.logger.error(e)); + }); + NotificationsService.watcher = watcher; + for (const event of ['add', 'change', 'unlink'] as const) { + watcher.on(event, (path) => { + void this.handleNotificationEvent(path, event).catch((e) => this.logger.error(e)); }); - - return NotificationsService.watcher; + } + await new Promise((resolve, reject) => { + watcher.once('error', reject); + watcher.once('ready', () => { + watcher.off('error', reject); + resolve(); + }); + }); + return watcher; } private isMissingFileError(error: unknown): error is NodeJS.ErrnoException { return error instanceof Error && 'code' in error && error.code === 'ENOENT'; } - private async handleNotificationAdd(path: string) { + private async handleNotificationEvent(path: string, event: NotificationFileEvent) { + const paths = this.paths(); + const type = Object.values(NotificationType).find((type) => dirname(path) === paths[type]); + if (!type) return; if (this.isHydratingOverview) { - this.pendingNotificationAdds.add(path); + const pending = this.pendingNotificationEvents.get(path); + this.pendingNotificationEvents.set( + path, + event === 'change' && pending === 'add' ? 'add' : event + ); return; } - await this.processNotificationAdd(path); + const { unread } = await this.synchronizeNotification(basename(path)); + if (type === NotificationType.UNREAD) { + if (event === 'add' && unread) { + await pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_ADDED, { notificationAdded: unread }); + } + void this.publishWarningsAndAlerts(); + } } - private async processNotificationAdd(path: string) { - // The path looks like /{notification base path}/{type}/{notification id} - const type = path.includes('/unread/') ? NotificationType.UNREAD : NotificationType.ARCHIVE; - // this.logger.debug(`Adding ${type} Notification: ${path}`); + private enqueueOverviewUpdate(action: () => Promise): Promise { + const update = this.overviewUpdate.catch(() => undefined).then(action); + this.overviewUpdate = update; + return update; + } - let notification: Notification; + private async readNotificationCopy(id: string, type: NotificationType) { try { - notification = await this.loadNotificationFile(path, NotificationType[type]); + return await this.loadNotificationFile(this.notificationPath(id, type), type); } catch (error) { - if (this.isMissingFileError(error)) { - this.logger.debug( - `[handleNotificationAdd] Notification disappeared before load: ${path}` - ); - return; - } + if (this.isMissingFileError(error)) return null; throw error; } + } - await this.recalculateOverview(); - - if (type === NotificationType.UNREAD) { - pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_ADDED, { - notificationAdded: notification, + private synchronizeNotification(id: string) { + return this.enqueueOverviewUpdate(async () => { + const [unread, archive] = await Promise.all([ + this.readNotificationCopy(id, NotificationType.UNREAD), + this.readNotificationCopy(id, NotificationType.ARCHIVE), + ]); + const changed = this.updateNotificationCounts(id, { + UNREAD: unread?.importance, + ARCHIVE: archive?.importance, }); - void this.publishWarningsAndAlerts(); + if (changed) await this.publishOverview(); + return { unread }; + }); + } + + private updateNotificationCounts(id: string, next: NotificationFileState): boolean { + const previous = this.notificationFiles.get(id) ?? {}; + let changed = false; + for (const type of Object.values(NotificationType)) { + const previousImportance = this.visibleImportance(previous, type); + const nextImportance = this.visibleImportance(next, type); + if (previousImportance === nextImportance) continue; + const counts = NotificationsService.overview[type.toLowerCase()]; + if (previousImportance) { + counts[previousImportance.toLowerCase()]--; + counts.total--; + } + if (nextImportance) this.increment(nextImportance, counts); + changed = true; } + if (next.UNREAD === undefined && next.ARCHIVE === undefined) { + this.notificationFiles.delete(id); + } else { + this.notificationFiles.set(id, next); + } + return changed; + } + + private visibleImportance(files: NotificationFileState, type: NotificationType) { + return type === NotificationType.UNREAD || files.UNREAD === undefined ? files[type] : undefined; } /** @@ -231,7 +280,7 @@ export class NotificationsService { return structuredClone(NotificationsService.overview); } - private publishOverview(overview = NotificationsService.overview) { + private publishOverview(overview = this.getOverview()) { return pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_OVERVIEW, { notificationsOverview: overview, }); @@ -257,28 +306,20 @@ export class NotificationsService { } public async recalculateOverview() { - const { errorOccurred, overview } = await this.refreshOverview(); - return { error: errorOccurred, overview: structuredClone(overview) }; + await this.waitForInitialization(); + return this.reconcileOverview(); } - private refreshOverview() { - if (this.pendingOverviewRefresh) { - return this.pendingOverviewRefresh; - } - - const refresh = this.overviewRefresh - .catch(() => undefined) - .then(async () => { - // Coalesce queued events, but refresh again for changes during an active snapshot. - this.pendingOverviewRefresh = null; - const snapshot = await this.buildOverviewSnapshot(); + private reconcileOverview() { + return this.enqueueOverviewUpdate(async () => { + const snapshot = await this.buildOverviewSnapshot(); + if (!snapshot.errorOccurred) { + this.notificationFiles = snapshot.files; NotificationsService.overview = snapshot.overview; await this.publishOverview(); - return snapshot; - }); - this.overviewRefresh = refresh; - this.pendingOverviewRefresh = refresh; - return refresh; + } + return { error: snapshot.errorOccurred, overview: this.getOverview() }; + }); } private async buildOverviewSnapshot() { @@ -296,29 +337,30 @@ export class NotificationsService { total: 0, }, }; - const seenPaths = new Set(); - - // todo - refactor this to be more memory efficient - // i.e. by using a lazy generator vs the current eager implementation - // - // recalculates stats for a particular notification type + const files = new Map(); const recalculate = async (type: NotificationType) => { - const ids = await this.listNotificationFiles(type); - ids.forEach((id) => seenPaths.add(id)); - const [notifications] = await this.loadNotificationsFromPaths(ids, { type }); - notifications.forEach((n) => this.increment(n.importance, overview[type.toLowerCase()])); + // Keep hidden archive copies in the index so removing unread can reveal them. + const paths = await this.listFilesInFolder(this.paths()[type]); + const [notifications, errors] = await this.loadNotificationsFromPaths(paths, { type }); + const failure = errors.find((error) => !this.isMissingFileError(error)); + if (failure) throw failure; + for (const notification of notifications) { + const copies = files.get(notification.id) ?? {}; + copies[type] = notification.importance; + files.set(notification.id, copies); + } }; - - const results = await batchProcess( - [NotificationType.ARCHIVE, NotificationType.UNREAD], - recalculate - ); - + const results = await batchProcess(Object.values(NotificationType), recalculate); if (results.errorOccurred) { - results.errors.forEach((e) => this.logger.error('[recalculateOverview] ' + e)); + results.errors.forEach((error) => this.logger.error('[recalculateOverview] ' + error)); } - - return { errorOccurred: results.errorOccurred, overview, seenPaths }; + for (const copies of files.values()) { + for (const type of Object.values(NotificationType)) { + const importance = this.visibleImportance(copies, type); + if (importance) this.increment(importance, overview[type.toLowerCase()]); + } + } + return { errorOccurred: results.errorOccurred, overview, files }; } /**------------------------------------------------------------------------ @@ -326,6 +368,7 @@ export class NotificationsService { *------------------------------------------------------------------------**/ public async createNotification(data: NotificationData): Promise { + await this.waitForInitialization(); const id: string = await this.makeNotificationId(data.title); const fileData = this.makeNotificationFileData(data); @@ -421,6 +464,7 @@ export class NotificationsService { *------------------------------------------------------------------------**/ public async deleteNotification({ id, type }: Pick) { + await this.waitForInitialization(); const path = this.notificationPath(id, type); // we don't want to update the overview stats if the deletion (unlink) fails @@ -429,7 +473,7 @@ export class NotificationsService { const notification = await this.loadNotificationFile(path, type); await unlink(path); - await this.recalculateOverview(); + await this.synchronizeNotification(id); if (type === NotificationType.UNREAD) { void this.publishWarningsAndAlerts(); } @@ -446,8 +490,14 @@ export class NotificationsService { * @remarks Ensures the notifications directory exists before emptying it */ public async deleteNotifications(type: NotificationType) { - await emptyDir(this.paths()[type]); - await this.recalculateOverview(); + await this.waitForInitialization(); + const folder = this.paths()[type]; + const ids = new Set(await readdir(folder)); + for (const [id, copies] of this.notificationFiles) { + if (copies[type] !== undefined) ids.add(id); + } + await emptyDir(folder); + await Promise.all([...ids].map((id) => this.synchronizeNotification(id))); if (type === NotificationType.UNREAD) { void this.publishWarningsAndAlerts(); } @@ -477,11 +527,12 @@ export class NotificationsService { const targetPath = this.notificationPath(notification.id, to); await rename(currentPath, targetPath); // A legacy archive copy may already exist, so rename need not emit an add event. - await this.recalculateOverview(); + await this.synchronizeNotification(notification.id); }; } public async archiveNotification({ id }: Pick): Promise { + await this.waitForInitialization(); const unreadPath = this.notificationPath(id, NotificationType.UNREAD); // We expect to only archive 'unread' notifications, but it's possible that the notification @@ -507,6 +558,7 @@ export class NotificationsService { } public async markAsUnread({ id }: Pick): Promise { + await this.waitForInitialization(); const archivePath = this.notificationPath(id, NotificationType.ARCHIVE); // the target notification might not be in the archive! if (!(await fileExists(archivePath))) { @@ -529,6 +581,7 @@ export class NotificationsService { } public async archiveAll(importance?: NotificationImportance) { + await this.waitForInitialization(); const { UNREAD } = this.paths(); if (!importance) { @@ -549,6 +602,7 @@ export class NotificationsService { } public async unarchiveAll(importance?: NotificationImportance) { + await this.waitForInitialization(); const { ARCHIVE } = this.paths(); if (!importance) { @@ -607,6 +661,7 @@ export class NotificationsService { * @returns An array of all notifications in the system. */ public async getNotifications(filters: NotificationFilter): Promise { + await this.waitForInitialization(); this.logger.verbose('Getting Notifications'); const { type = NotificationType.UNREAD } = filters; @@ -931,6 +986,7 @@ export class NotificationsService { *------------------------------------------------------------------------**/ private async loadUnreadNotifications(): Promise { + await this.waitForInitialization(); const { UNREAD } = this.paths(); const files = await this.listFilesInFolder(UNREAD); const [notifications] = await this.loadNotificationsFromPaths(files, { From fdef73bba0995652a8c29cd18cda9dd3d5bec191 Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sun, 25 Oct 2026 12:23:17 -0400 Subject: [PATCH 4/8] test(notifications): add native notifier appliance checks --- api/scripts/qa/README.md | 52 +++++ api/scripts/qa/notification-counter.mjs | 197 ++++++++++++++++++ .../loadNotificationsFile.test.ts | 64 +----- 3 files changed, 258 insertions(+), 55 deletions(-) create mode 100644 api/scripts/qa/README.md create mode 100644 api/scripts/qa/notification-counter.mjs diff --git a/api/scripts/qa/README.md b/api/scripts/qa/README.md new file mode 100644 index 0000000000..95a203fb77 --- /dev/null +++ b/api/scripts/qa/README.md @@ -0,0 +1,52 @@ +# Notification counter appliance test + +`notification-counter.mjs` tests the installed API against Unraid's native +`/usr/local/emhttp/webGui/scripts/notify`. Run it inside a disposable, managed +QA guest after installing the production-shaped plugin for the selected commit. +The guest must have `/boot/config/qa-harness-ok` and Node.js with native `fetch`. + +The QA runner must establish the source commit and package digest before +installation. Extract the API entrypoint from that same package and calculate +its SHA-256. Supply that expected hash to the guest; do not calculate the +expected value from the guest's installed file. The script compares the +installed entrypoint with this hash and records the native notifier's hash. +This comparison covers the entrypoint. The runner remains responsible for +verifying the complete installed package and its revision. + +Supply these environment variables through the managed guest execution path: + +- `QA_GRAPHQL_URL`: authenticated local GraphQL endpoint for this guest. +- `QA_API_KEY`: temporary key with notification read, update, and delete permissions. +- `QA_API_ENTRYPOINT`: absolute installed API entrypoint path from the package. +- `QA_API_SHA256`: expected SHA-256 from the pinned package. +- `QA_SOURCE_REVISION`: full 40-character source commit for that package. +- `QA_REPORT_PATH`: fresh absolute path for the private JSON receipt. + +Run `node /absolute/guest/path/notification-counter.mjs`. Use the QA VM +provider's current lifecycle and execution contract. Do not run this script on +a production server. The runner must restore and release the guest even when +the test fails, and remove its temporary API key. + +The script uses `notify -i " 1"` to enable browser-file delivery only. +It does not change notification settings or invoke email and notification agents. +It creates uniquely named files and removes only those files in `finally`. +It verifies: + +- All three severities through the actual notifier, including both unread and + archive copies of each notification. +- Native archive actions, API restore, API archive over an existing copy, and API deletion. +- External severity changes and unread deletion that reveals the archive's severity. +- Native archive-only delivery and a burst of 24 native file writes. +- Matching cached counters, visible GraphQL lists, and independent disk counts + after every step and after cleanup. + +The script never calls `recalculateOverview`, so a watcher failure cannot be +hidden by a manual refresh. Each check allows 15 seconds for watcher delivery. +It stops if notification ground truth is ambiguous or a list reaches its +10,000-item cap. Existing notifications must have a supported importance value. +The receipt contains hashes, source revision, fixture counts, and check names; +it excludes credentials, the endpoint, and existing notification contents. + +A syntax check is not appliance acceptance. Only a receipt with `status=passed` +and `cleanup=complete`, plus the runner's package and VM lifecycle evidence, +proves that this appliance test passed. diff --git a/api/scripts/qa/notification-counter.mjs b/api/scripts/qa/notification-counter.mjs new file mode 100644 index 0000000000..7122fc154d --- /dev/null +++ b/api/scripts/qa/notification-counter.mjs @@ -0,0 +1,197 @@ +#!/usr/bin/env node +// Run inside a managed, disposable Unraid QA guest after installing the pinned plugin. +import assert from 'node:assert/strict'; +import { createHash, randomUUID } from 'node:crypto'; +import { execFileSync } from 'node:child_process'; +import { access, readFile, readdir, unlink, writeFile } from 'node:fs/promises'; +import { isAbsolute, join } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; + +const required = (name) => { + assert(process.env[name], `Missing ${name}`); + return process.env[name]; +}; +const endpoint = required('QA_GRAPHQL_URL'); +const key = required('QA_API_KEY'); +const entrypoint = required('QA_API_ENTRYPOINT'); +const expectedHash = required('QA_API_SHA256'); +const revision = required('QA_SOURCE_REVISION'); +const reportPath = required('QA_REPORT_PATH'); +assert(/^[a-f0-9]{64}$/.test(expectedHash), 'Expected packaged API SHA-256'); +assert(/^[a-f0-9]{40}$/.test(revision), 'Expected exact source commit'); +assert(isAbsolute(entrypoint) && isAbsolute(reportPath), 'Use absolute guest paths'); +const notifier = '/usr/local/emhttp/webGui/scripts/notify'; +const prefix = `qa1859_${randomUUID().replaceAll('-', '')}`; +const checks = []; +const report = { revision, checks, status: 'running' }; +let notificationPath; +let stage = 'preflight'; +const hash = (bytes) => createHash('sha256').update(bytes).digest('hex'); +const counts = '{info warning alert total}'; +const overviewFields = `unread ${counts} archive ${counts}`; +const query = `query { notifications { + overview { ${overviewFields} } + unread: list(filter: {type: UNREAD, offset: 0, limit: 10000}) {id title importance} + archive: list(filter: {type: ARCHIVE, offset: 0, limit: 10000}) {id title importance} +} }`; + +async function graphql(query, variables = {}) { + const response = await fetch(endpoint, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-api-key': key }, + body: JSON.stringify({ query, variables }), + signal: AbortSignal.timeout(10000), + }); + assert.equal(response.status, 200, `GraphQL HTTP ${response.status}`); + const body = await response.json(); + assert(!body.errors?.length, `GraphQL errors: ${body.errors?.map((e) => e.message).join('; ')}`); + return body.data; +} + +const emptyCounts = () => ({ info: 0, warning: 0, alert: 0, total: 0 }); +function summarize(notifications) { + const result = emptyCounts(); + for (const notification of notifications) { + const importance = notification.importance.toLowerCase(); + assert(Object.hasOwn(result, importance) && importance !== 'total', 'Unknown importance'); + result[importance]++; + result.total++; + } + return result; +} +const localId = (id) => id.split(':').at(-1); +async function diskState() { + const state = {}; + for (const type of ['unread', 'archive']) { + state[type] = []; + for (const id of await readdir(join(notificationPath, type))) { + if (!id.endsWith('.notify')) continue; + const text = await readFile(join(notificationPath, type, id), 'utf8'); + const severity = text.match(/^importance\s*=\s*"?(normal|warning|alert)"?\s*$/m)?.[1]; + assert(severity, 'Disk fixture has unsupported importance; cannot establish ground truth'); + state[type].push({ id, importance: severity === 'normal' ? 'INFO' : severity.toUpperCase() }); + } + } + const unreadIds = new Set(state.unread.map((n) => n.id)); + state.archive = state.archive.filter((n) => !unreadIds.has(n.id)); + return state; +} + +async function verify(label, fixtureCounts) { + stage = label; + const deadline = Date.now() + 15000; + let lastError; + do { + try { + const disk = await diskState(); + const { notifications } = await graphql(query); + const fixtureOverview = {}; + for (const type of ['unread', 'archive']) { + const list = notifications[type]; + assert(list.length < 10000, 'List cap reached; cannot establish complete ground truth'); + assert.deepEqual(list.map((n) => localId(n.id)).sort(), disk[type].map((n) => n.id).sort()); + assert.deepEqual(notifications.overview[type], summarize(disk[type])); + assert.deepEqual(notifications.overview[type], summarize(list)); + fixtureOverview[type] = summarize(list.filter((n) => n.title.startsWith(prefix))); + assert.equal(fixtureOverview[type].total, fixtureCounts[type]); + } + checks.push({ label, fixtureOverview, diskMatchesListsAndCounters: true }); + console.log(`PASS ${label}`); + return notifications; + } catch (error) { + lastError = error; + await delay(150); + } + } while (Date.now() < deadline); + throw new Error(`${label}: ${lastError.message}`); +} + +function notify(event, severity, archiveOnly = false) { + execFileSync(notifier, [ + '-e', event, '-s', 'Notification counter QA', '-d', 'Disposable notification fixture', + '-i', `${severity} 1`, '-x', ...(archiveOnly ? ['-b'] : []), + ], { timeout: 10000, stdio: 'pipe' }); + return `${event}_ticket.notify`; +} +async function mutation(name, id) { + return graphql(`mutation($id: PrefixedID!) { ${name}(id: $id) { id } }`, { id }); +} +async function cleanup() { + if (!notificationPath) return; + for (const type of ['unread', 'archive']) { + for (const id of await readdir(join(notificationPath, type))) { + if (id.startsWith(prefix) && id.endsWith('.notify')) { + await unlink(join(notificationPath, type, id)); + } + } + } + await verify('fixture cleanup', { unread: 0, archive: 0 }); +} + +// Write a fresh private report before mutations; never overwrite another run's evidence. +await writeFile(reportPath, JSON.stringify(report), { flag: 'wx', mode: 0o600 }); +try { + await access('/boot/config/qa-harness-ok'); + const apiBytes = await readFile(entrypoint); + report.apiSha256 = hash(apiBytes); + assert.equal(report.apiSha256, expectedHash, 'Installed API differs from pinned package'); + const script = await readFile(notifier, 'utf8'); + report.notifierSha256 = hash(script); + assert(script.includes('$overrule') && script.includes('strtok'), 'Notifier delivery override unavailable'); + // The native parser resolves defaults and persistent settings exactly as notify does. + const config = JSON.parse(execFileSync('/usr/bin/php', ['-r', + '$docroot="/usr/local/emhttp"; require_once "$docroot/webGui/include/Wrappers.php"; ' + + '$c=parse_plugin_cfg("dynamix",true); echo json_encode($c["notify"]["path"] ?? "/tmp/notifications");', + ], { timeout: 10000, encoding: 'utf8' })); + assert(typeof config === 'string' && isAbsolute(config), 'Invalid configured notification path'); + notificationPath = config; + await verify('baseline', { unread: 0, archive: 0 }); + const ids = ['normal', 'warning', 'alert'].map((severity) => notify(`${prefix}_${severity}`, severity)); + for (const id of ids) { + await access(join(notificationPath, 'unread', id)); + await access(join(notificationPath, 'archive', id)); + } + const created = await verify('native dual writes, all severities', { unread: 3, archive: 0 }); + assert.deepEqual(summarize(created.unread.filter((n) => n.title.startsWith(prefix))), + { info: 1, warning: 1, alert: 1, total: 3 }); + execFileSync(notifier, ['archive', ids[0]], { timeout: 10000, stdio: 'pipe' }); + await verify('native archive reveals existing archive copy', { unread: 2, archive: 1 }); + await mutation('unreadNotification', ids[0]); + await verify('API restore', { unread: 3, archive: 0 }); + await mutation('archiveNotification', ids[1]); + await verify('API archive replaces duplicate', { unread: 2, archive: 1 }); + await graphql(`mutation($id: PrefixedID!) { + deleteNotification(id: $id, type: ARCHIVE) { ${overviewFields} } + }`, { id: ids[1] }); + await verify('API archive deletion', { unread: 2, archive: 0 }); + const alertPath = join(notificationPath, 'unread', ids[2]); + const text = await readFile(alertPath, 'utf8'); + await writeFile(alertPath, text.replace(/^importance=.*$/m, 'importance="warning"')); + const changed = await verify('external severity change', { unread: 2, archive: 0 }); + assert.deepEqual(summarize(changed.unread.filter((n) => n.title.startsWith(prefix))), + { info: 1, warning: 1, alert: 0, total: 2 }); + await unlink(alertPath); + await verify('external unread deletion reveals archive severity', { unread: 1, archive: 1 }); + notify(`${prefix}_archive_only`, 'normal', true); + await verify('native archive-only write', { unread: 1, archive: 2 }); + for (let i = 0; i < 12; i++) notify(`${prefix}_burst_${i}`, ['normal', 'warning', 'alert'][i % 3]); + await verify('native burst of 24 file writes', { unread: 13, archive: 2 }); + report.status = 'passed'; +} catch (error) { + report.status = 'failed'; + // Never include credentials, endpoint, or existing notification contents in the receipt. + report.failedStage = stage; + report.errorType = error.name; + process.exitCode = 1; +} finally { + try { + await cleanup(); + report.cleanup = 'complete'; + } catch { + report.cleanup = 'failed'; + report.status = 'failed'; + process.exitCode = 1; + } + await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, { mode: 0o600 }); + console.log(`Result: ${report.status}; cleanup: ${report.cleanup}`); +} diff --git a/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts b/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts index 09b3663e45..be5c43f4c4 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts @@ -15,14 +15,16 @@ import { } from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; -const { mockWatch } = vi.hoisted(() => { - const watcher = { - on: vi.fn().mockReturnThis(), - close: vi.fn().mockResolvedValue(undefined), - }; - +const { mockWatch } = await vi.hoisted(async () => { + const { EventEmitter } = await import('node:events'); return { - mockWatch: vi.fn(() => watcher), + mockWatch: vi.fn(() => { + const watcher = Object.assign(new EventEmitter(), { + close: vi.fn().mockResolvedValue(undefined), + }); + setImmediate(() => watcher.emit('ready')); + return watcher; + }), }; }); @@ -129,54 +131,6 @@ describe('NotificationsService - loadNotificationFile (minimal mocks)', () => { ); }); - it('replays buffered add events after overview hydration', async () => { - const bufferedPath = `${testNotificationsDir}/unread/buffered.notify`; - const hydratedService = createNotificationsService(); - const processNotificationAdd = vi.fn().mockResolvedValue(undefined); - const handleNotificationAdd = ( - Reflect.get(hydratedService, 'handleNotificationAdd') as (path: string) => Promise - ).bind(hydratedService); - - Reflect.set( - hydratedService, - 'ensureNotificationDirectories', - vi.fn().mockResolvedValue(undefined) - ); - Reflect.set(hydratedService, 'publishOverview', vi.fn().mockResolvedValue(undefined)); - Reflect.set(hydratedService, 'processNotificationAdd', processNotificationAdd); - Reflect.set( - hydratedService, - 'getNotificationsWatcher', - vi.fn().mockImplementation(async () => { - await handleNotificationAdd(bufferedPath); - return { - close: vi.fn().mockResolvedValue(undefined), - on: vi.fn().mockReturnThis(), - }; - }) - ); - Reflect.set( - hydratedService, - 'buildOverviewSnapshot', - vi.fn().mockResolvedValue({ - errorOccurred: false, - overview: { - unread: { alert: 0, info: 0, warning: 0, total: 0 }, - archive: { alert: 0, info: 0, warning: 0, total: 0 }, - }, - seenPaths: new Set(), - }) - ); - - await Reflect.get(hydratedService, 'initializeNotificationsState').call( - hydratedService, - testNotificationsDir, - true - ); - - expect(processNotificationAdd).toHaveBeenCalledWith(bufferedPath); - }); - it('should load and validate a valid notification file', async () => { const mockFileContent = `timestamp=1609459200 event=Test Event From 29a80c18a35deb9df2b038962f571974d6d47115 Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sun, 25 Oct 2026 13:14:11 -0400 Subject: [PATCH 5/8] fix(notifications): recover watcher initialization failures --- api/scripts/qa/README.md | 28 +++ api/scripts/qa/notification-counter.mjs | 226 +++++++++++++++++- .../loadNotificationsFile.test.ts | 31 +++ .../notifications.overview.spec.ts | 76 ++++++ .../notifications/notifications.service.ts | 31 ++- 5 files changed, 381 insertions(+), 11 deletions(-) diff --git a/api/scripts/qa/README.md b/api/scripts/qa/README.md index 95a203fb77..ec56699d84 100644 --- a/api/scripts/qa/README.md +++ b/api/scripts/qa/README.md @@ -50,3 +50,31 @@ it excludes credentials, the endpoint, and existing notification contents. A syntax check is not appliance acceptance. Only a receipt with `status=passed` and `cleanup=complete`, plus the runner's package and VM lifecycle evidence, proves that this appliance test passed. + +## Adversarial appliance pass + +Set `QA_ADVERSARIAL=true` and `QA_ALLOW_SERVICE_RESTART=true` to run the extended +pass. This requires a disposable guest, an isolated empty notification directory, +and native notification settings `normal=1`, `warning=1`, and `alert=1`. The runner +must prepare those settings before restarting the API, preserve the original +configuration, and restore it afterward. The temporary API key also needs +notification creation permission. With the current CLI, use an alphanumeric key +name, an explicit `GUEST` role, and the four explicit notification permissions. + +The extended pass checks API creation through the installed native script, +three authenticated GraphQL WebSocket subscriptions, content updates without +count changes, selected-ID and severity-filtered bulk archive/restore, all-item +archive/restore/deletion, persisted counts after service restart, subscription +reconnection, and recovery from a real startup directory failure without +another process restart. + +It also temporarily lowers the guest's inotify watch limit, creates a watched +subdirectory, requires a logged `ENOSPC`, restores the original limit in +`finally`, and proves that the same API process still counts native writes. +The runner must restore and release the VM on every outcome. This fault test +must never run on a shared or production guest. + +The receipt separately records whether the ID returned by native API creation +matches the actual disk record. A mismatch produces `passed_with_findings`, +which is not a clean regression verdict. The runner must report that finding +and establish whether it predates the candidate before deciding its scope. diff --git a/api/scripts/qa/notification-counter.mjs b/api/scripts/qa/notification-counter.mjs index 7122fc154d..541365e4b6 100644 --- a/api/scripts/qa/notification-counter.mjs +++ b/api/scripts/qa/notification-counter.mjs @@ -3,7 +3,8 @@ import assert from 'node:assert/strict'; import { createHash, randomUUID } from 'node:crypto'; import { execFileSync } from 'node:child_process'; -import { access, readFile, readdir, unlink, writeFile } from 'node:fs/promises'; +import { access, mkdir, readFile, readdir, rename, rm, unlink, writeFile } from 'node:fs/promises'; +import { createRequire } from 'node:module'; import { isAbsolute, join } from 'node:path'; import { setTimeout as delay } from 'node:timers/promises'; @@ -31,8 +32,8 @@ const counts = '{info warning alert total}'; const overviewFields = `unread ${counts} archive ${counts}`; const query = `query { notifications { overview { ${overviewFields} } - unread: list(filter: {type: UNREAD, offset: 0, limit: 10000}) {id title importance} - archive: list(filter: {type: ARCHIVE, offset: 0, limit: 10000}) {id title importance} + unread: list(filter: {type: UNREAD, offset: 0, limit: 10000}) {id title subject description importance link} + archive: list(filter: {type: ARCHIVE, offset: 0, limit: 10000}) {id title subject description importance link} } }`; async function graphql(query, variables = {}) { @@ -128,6 +129,218 @@ async function cleanup() { await verify('fixture cleanup', { unread: 0, archive: 0 }); } +async function waitUntil(predicate, label, timeout = 15000) { + const deadline = Date.now() + timeout; + while (Date.now() < deadline) { + if (await predicate()) return; + await delay(100); + } + throw new Error(`Timed out: ${label}`); +} + +function apiProcess() { + const require = createRequire(entrypoint); + const list = JSON.parse(execFileSync(process.execPath, [require.resolve('pm2/bin/pm2'), 'jlist'], { + encoding: 'utf8', timeout: 10000, + env: { ...process.env, PM2_HOME: '/var/log/.pm2', PM2_DISCRETE_MODE: 'true' }, + stdio: ['ignore', 'pipe', 'pipe'], + })); + const app = list.find((app) => app.name === 'unraid-api' && app.pm2_env.pm_exec_path === entrypoint); + assert(app?.pid > 0, 'API process not running'); + return { pid: app.pid, restarts: app.pm2_env.restart_time }; +} + +async function observeSubscriptions() { + const require = createRequire(entrypoint); + const WebSocket = require('ws'); + const socket = new WebSocket(endpoint.replace(/^http/, 'ws'), 'graphql-transport-ws', { + headers: { 'x-api-key': key }, + }); + const messages = []; + let failure = false; + socket.on('error', () => { failure = true; }); + socket.on('message', (bytes) => { + try { messages.push(JSON.parse(bytes.toString())); } catch { failure = true; } + }); + socket.on('open', () => socket.send(JSON.stringify({ + type: 'connection_init', payload: { 'x-api-key': key }, + }))); + const until = async (predicate, label) => waitUntil(() => { + assert(!failure && !messages.some((m) => m.type === 'error' || m.payload?.errors), 'Subscription failed'); + return predicate(); + }, label); + try { + await until(() => messages.some((m) => m.type === 'connection_ack'), 'WebSocket acknowledgement'); + const fields = { + added: 'notificationAdded {id title subject importance}', + overview: `notificationsOverview {${overviewFields}}`, + warnings: 'notificationsWarningsAndAlerts {id title subject importance}', + }; + for (const [id, field] of Object.entries(fields)) { + socket.send(JSON.stringify({ id, type: 'subscribe', payload: { query: `subscription {${field}}` } })); + } + return { + messages, + until, + close: async () => { + socket.close(); + try { await waitUntil(() => socket.readyState === WebSocket.CLOSED, 'WebSocket close', 2000); } + finally { if (socket.readyState !== WebSocket.CLOSED) socket.terminate(); } + }, + }; + } catch (error) { + socket.terminate(); + throw error; + } +} + +async function adversarial() { + stage = 'adversarial preflight'; + assert.equal(process.env.QA_ALLOW_SERVICE_RESTART, 'true', 'Explicit disposable-guest restart permission required'); + const config = JSON.parse(execFileSync('/usr/bin/php', ['-r', + '$docroot="/usr/local/emhttp"; require_once "$docroot/webGui/include/Wrappers.php"; ' + + '$c=parse_plugin_cfg("dynamix",true); echo json_encode($c["notify"]);', + ], { encoding: 'utf8', timeout: 10000 })); + for (const severity of ['normal', 'warning', 'alert']) { + assert.equal(Number(config[severity]), 1, 'Native API creation requires browser-only notification settings'); + } + const baseline = await graphql(query); + assert.equal(baseline.notifications.unread.length + baseline.notifications.archive.length, 0, + 'Adversarial bulk tests require an isolated empty notification directory'); + let observer = await observeSubscriptions(); + try { + stage = 'native API creation and subscriptions'; + const title = `${prefix}_api`; + const { createNotification: created } = await graphql(`mutation($input: NotificationData!) { + createNotification(input: $input) {id title subject description importance link} + }`, { input: { title, subject: 'Quoted "subject" = value', description: 'Notification QA & content', + importance: 'WARNING', link: '/Settings/Notifications' } }); + const state = await verify('API creation through installed native notifier', { unread: 1, archive: 0 }); + const actual = state.unread.find((n) => n.title === title); + assert(actual, 'Created notification absent from list'); + // A native archive copy proves the API did not take its local-write fallback. + await access(join(notificationPath, 'archive', localId(actual.id))); + assert.equal(created.subject, 'Quoted "subject" = value'); + for (const field of ['title', 'subject', 'description', 'importance', 'link']) { + assert.equal(actual[field], created[field], 'Native file metadata differs from API creation response'); + } + report.nativeCreateReturnedIdMatchesDisk = localId(created.id) === localId(actual.id); + await observer.until(() => observer.messages.some((m) => m.id === 'added' + && m.payload?.data?.notificationAdded?.title === title), 'live notificationAdded'); + await observer.until(() => observer.messages.some((m) => m.id === 'overview' + && m.payload?.data?.notificationsOverview?.unread.warning === 1), 'live notificationsOverview'); + await observer.until(() => observer.messages.some((m) => m.id === 'warnings' + && m.payload?.data?.notificationsWarningsAndAlerts?.some((n) => n.title === title)), 'live warning subscription'); + checks.push({ label: 'three authenticated live GraphQL subscriptions', passed: true }); + const original = await readFile(join(notificationPath, 'unread', localId(actual.id)), 'utf8'); + await writeFile(join(notificationPath, 'unread', localId(actual.id)), + original.replace(/^subject=.*$/m, 'subject="Changed content"')); + await observer.until(() => observer.messages.some((m) => m.id === 'warnings' + && m.payload?.data?.notificationsWarningsAndAlerts?.some((n) => n.title === title + && n.subject === 'Changed content')), 'warning content update without count change'); + checks.push({ label: 'live content update at unchanged severity', passed: true }); + await mutation('archiveNotification', actual.id); + await verify('API-created notification can be archived by its listed ID', { unread: 0, archive: 1 }); + await observer.until(() => observer.messages.some((m) => m.id === 'warnings' + && m.payload?.data?.notificationsWarningsAndAlerts?.length === 0), 'warning removal subscription'); + await cleanup(); + + stage = 'bulk regression'; + const ids = ['normal', 'warning', 'alert'].map((s) => notify(`${prefix}_bulk_${s}`, s)); + await verify('bulk seed', { unread: 3, archive: 0 }); + await observer.until(() => ids.every((id) => observer.messages.some((m) => m.id === 'added' + && localId(m.payload?.data?.notificationAdded?.id ?? '') === id)), 'native bulk add events'); + for (const id of ids) { + const additions = observer.messages.filter((m) => m.id === 'added' + && localId(m.payload?.data?.notificationAdded?.id ?? '') === id); + assert.equal(additions.length, 1, 'Native add was lost or published more than once'); + } + checks.push({ label: 'one live notificationAdded per new native file', passed: true }); + notify(`${prefix}_bulk_normal`, 'normal'); + await verify('repeated native ticket creates no duplicate', { unread: 3, archive: 0 }); + await graphql(`mutation($ids: [PrefixedID!]!) {archiveNotifications(ids: $ids) {${overviewFields}}}`, { ids: ids.slice(0, 2) }); + await verify('bulk archive selected IDs', { unread: 1, archive: 2 }); + await graphql(`mutation($ids: [PrefixedID!]!) {unarchiveNotifications(ids: $ids) {${overviewFields}}}`, { ids: ids.slice(0, 2) }); + await verify('bulk restore selected IDs', { unread: 3, archive: 0 }); + await graphql(`mutation {archiveAll(importance: WARNING) {${overviewFields}}}`); + await verify('bulk archive by severity', { unread: 2, archive: 1 }); + await graphql(`mutation {unarchiveAll(importance: WARNING) {${overviewFields}}}`); + await verify('bulk restore by severity', { unread: 3, archive: 0 }); + await graphql(`mutation {archiveAll {${overviewFields}}}`); + await verify('archive all', { unread: 0, archive: 3 }); + await graphql(`mutation {unarchiveAll {${overviewFields}}}`); + await verify('restore all', { unread: 3, archive: 0 }); + await graphql(`mutation {archiveAll {${overviewFields}}}`); + await graphql(`mutation {deleteArchivedNotifications {${overviewFields}}}`); + await verify('delete all archived', { unread: 0, archive: 0 }); + await observer.close(); + observer = null; + + stage = 'restart persistence'; + const persistent = notify(`${prefix}_restart`, 'alert'); + await verify('before service restart', { unread: 1, archive: 0 }); + const before = apiProcess(); + execFileSync('/etc/rc.d/rc.unraid-api', ['restart'], { timeout: 60000, stdio: 'pipe' }); + await verify('rehydrates persisted native copies after restart', { unread: 1, archive: 0 }); + assert.notEqual(apiProcess().pid, before.pid, 'API process did not restart'); + observer = await observeSubscriptions(); + const reconnect = notify(`${prefix}_reconnect`, 'warning'); + await verify('new native write after restart', { unread: 2, archive: 0 }); + await observer.until(() => observer.messages.some((m) => m.id === 'added' + && localId(m.payload?.data?.notificationAdded?.id ?? '') === reconnect), 'subscription after restart'); + checks.push({ label: 'authenticated live subscription after restart', passed: true }); + await observer.close(); + observer = null; + + stage = 'startup filesystem failure'; + const unread = join(notificationPath, 'unread'); + const saved = join(notificationPath, `${prefix}_saved_unread`); + await rename(unread, saved); + await writeFile(unread, 'blocked directory'); + try { + execFileSync('/etc/rc.d/rc.unraid-api', ['restart'], { timeout: 60000, stdio: 'pipe' }); + const failedProcess = apiProcess(); + const response = await fetch(endpoint, { method: 'POST', + headers: { 'content-type': 'application/json', 'x-api-key': key }, + body: JSON.stringify({ query }), signal: AbortSignal.timeout(10000) }); + const body = await response.json(); + assert(body.errors?.length, 'Invalid directory did not expose an initialization error'); + await unlink(unread); + await rename(saved, unread); + await verify('recovers startup directory failure without another restart', { unread: 2, archive: 0 }); + assert.deepEqual(apiProcess(), failedProcess, 'Process restarted instead of recovering'); + } finally { + try { await access(saved); await rm(unread, { recursive: true, force: true }); await rename(saved, unread); } + catch (error) { if (error.code !== 'ENOENT') throw error; } + } + await mutation('archiveNotification', persistent); + await verify('mutation after initialization recovery', { unread: 1, archive: 1 }); + + stage = 'runtime native watcher error'; + const limitPath = '/proc/sys/fs/inotify/max_user_watches'; + const originalLimit = await readFile(limitPath, 'utf8'); + const logPath = '/var/log/graphql-api.log'; + const oldLogLength = (await readFile(logPath)).length; + const running = apiProcess(); + const faultDirectory = join(notificationPath, `${prefix}_watch_fault`); + try { + await writeFile(limitPath, '1\n'); + await mkdir(faultDirectory); + await waitUntil(async () => (await readFile(logPath, 'utf8')).slice(oldLogLength).includes('ENOSPC'), + 'actual inotify resource exhaustion'); + } finally { + await writeFile(limitPath, originalLimit); + await rm(faultDirectory, { recursive: true, force: true }); + } + assert.deepEqual(apiProcess(), running, 'Watcher error restarted the API process'); + notify(`${prefix}_after_watch_error`, 'normal'); + await verify('counts follow real writes after logged ENOSPC without process restart', { unread: 2, archive: 1 }); + checks.push({ label: 'native inotify error logged; process and counts remain operational', passed: true }); + } finally { + await observer?.close(); + } +} + // Write a fresh private report before mutations; never overwrite another run's evidence. await writeFile(reportPath, JSON.stringify(report), { flag: 'wx', mode: 0o600 }); try { @@ -176,7 +389,12 @@ try { await verify('native archive-only write', { unread: 1, archive: 2 }); for (let i = 0; i < 12; i++) notify(`${prefix}_burst_${i}`, ['normal', 'warning', 'alert'][i % 3]); await verify('native burst of 24 file writes', { unread: 13, archive: 2 }); - report.status = 'passed'; + if (process.env.QA_ADVERSARIAL === 'true') { + await cleanup(); + await adversarial(); + } + report.status = report.nativeCreateReturnedIdMatchesDisk === false ? 'passed_with_findings' : 'passed'; + if (report.status === 'passed_with_findings') process.exitCode = 1; } catch (error) { report.status = 'failed'; // Never include credentials, endpoint, or existing notification contents in the receipt. diff --git a/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts b/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts index be5c43f4c4..217db37827 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts @@ -2,6 +2,7 @@ import { ConfigService } from '@nestjs/config'; import { existsSync, mkdirSync, rmSync, writeFileSync } from 'fs'; +import { EventEmitter } from 'node:events'; import { tmpdir } from 'os'; import { join } from 'path'; @@ -131,6 +132,36 @@ describe('NotificationsService - loadNotificationFile (minimal mocks)', () => { ); }); + it('closes a watcher that fails before ready and retries on the next public read', async () => { + await Reflect.get(NotificationsService, 'watcher').close(); + Reflect.set(NotificationsService, 'watcher', null); + const failedWatcher = Object.assign(new EventEmitter(), { + close: vi.fn().mockResolvedValue(undefined), + }); + mockWatch.mockImplementationOnce(() => { + setImmediate(() => failedWatcher.emit('error', new Error('watch setup failed'))); + return failedWatcher; + }); + const recoveringService = createNotificationsService(); + await expect(Reflect.get(recoveringService, 'initialization')).rejects.toThrow(); + expect(failedWatcher.close).toHaveBeenCalledOnce(); + expect(Reflect.get(NotificationsService, 'watcher')).toBeNull(); + const attempts = mockWatch.mock.calls.length; + await expect( + Promise.all( + [0, 1].map(() => + recoveringService.getNotifications({ + type: NotificationType.UNREAD, + offset: 0, + limit: 10, + }) + ) + ) + ).resolves.toEqual([[], []]); + expect(mockWatch).toHaveBeenCalledTimes(attempts + 1); + expect(Reflect.get(NotificationsService, 'watcher')).not.toBe(failedWatcher); + }); + it('should load and validate a valid notification file', async () => { const mockFileContent = `timestamp=1609459200 event=Test Event diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts index 9648611ffd..61e081bc66 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts @@ -146,6 +146,33 @@ describe.sequential('Notification filesystem and subscription integration', () = await expectCounts(3, 3); }); + it('keeps handling real file changes after a runtime watcher error', async () => { + await start(); + expect(() => watcher.emit('error', new Error('watch resource exhausted'))).not.toThrow(); + await write(UNREAD, 'after-error.notify', 'warning'); + await expectEvent('add', UNREAD, 'after-error.notify'); + await expectCounts(1, 0); + await service.archiveNotification({ id: 'after-error.notify' }); + await expectCounts(0, 1); + }); + + it('recovers a failed directory initialization on the next public request', async () => { + const blocked = join(basePath, 'blocked'); + await writeFile(blocked, 'not a directory'); + config.set('store.dynamix.notify.path', blocked); + service = new NotificationsService(config); + await expect(Reflect.get(service, 'initialization')).rejects.toThrow(); + await unlink(blocked); + await expect(service.getNotifications({ type: UNREAD, offset: 0, limit: 10 })).resolves.toEqual( + [] + ); + await writeFile( + join(blocked, 'unread', 'recovered.notify'), + 'timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=alert\n' + ); + await vi.waitFor(() => expect(service.getOverview().unread.alert).toBe(1)); + }); + it.each([UNREAD, ARCHIVE])( 'handles legacy files written %s first without recounting', async (first) => { @@ -502,6 +529,55 @@ describe.sequential('Notification filesystem and subscription integration', () = await expectCounts(0, 0); }); + it('recovers a failed startup replay on the next public request', async () => { + const prototype = NotificationsService.prototype; + const snapshot = Reflect.get(prototype, 'buildOverviewSnapshot'); + const handler = Reflect.get(prototype, 'handleNotificationEvent'); + const captured = Promise.withResolvers(); + const release = Promise.withResolvers(); + let failReplay = true; + Reflect.set(prototype, 'buildOverviewSnapshot', async function (this: NotificationsService) { + const result = await snapshot.call(this); + captured.resolve(); + await release.promise; + return result; + }); + Reflect.set( + prototype, + 'handleNotificationEvent', + async function (this: NotificationsService, path: string, event: string) { + if ( + failReplay && + path.endsWith('/replay.notify') && + !Reflect.get(this, 'isHydratingOverview') + ) { + failReplay = false; + throw new Error('replay read failed'); + } + return handler.call(this, path, event); + } + ); + const rejected = expect(start()).rejects.toThrow(); + try { + await captured.promise; + const activeWatcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + const added = new Promise((resolve) => activeWatcher.once('add', () => resolve())); + await write(UNREAD, 'replay.notify', 'alert'); + await added; + release.resolve(); + await rejected; + expect(failReplay).toBe(false); + await expect( + service.getNotifications({ type: UNREAD, offset: 0, limit: 10 }) + ).resolves.toHaveLength(1); + await expectCounts(1, 0); + } finally { + release.resolve(); + Reflect.set(prototype, 'buildOverviewSnapshot', snapshot); + Reflect.set(prototype, 'handleNotificationEvent', handler); + } + }); + it('returns matching lists and counts through actual GraphQL queries and mutations', async () => { await write(UNREAD, 'graphql.notify', 'warning'); await write(ARCHIVE, 'graphql.notify', 'warning'); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts index c580f7a63c..ec6a30895b 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts @@ -46,6 +46,7 @@ export class NotificationsService { */ private path: string | null = null; private initialization: Promise | null = null; + private initializationFailed = false; private overviewUpdate: Promise = Promise.resolve(); private notificationFiles = new Map(); private isHydratingOverview = false; @@ -101,6 +102,9 @@ export class NotificationsService { private async waitForInitialization() { this.paths(); + if (this.initializationFailed) { + this.initialization = this.initializeNotificationsState(this.getConfiguredPath(), true); + } await this.initialization; } @@ -109,6 +113,7 @@ export class NotificationsService { } private initializeNotificationsState(basePath: string, recreate = false) { + this.initializationFailed = false; const initialize = async () => { await this.ensureNotificationDirectories(basePath); this.pendingNotificationEvents.clear(); @@ -137,7 +142,12 @@ export class NotificationsService { }; const previousInitialization = this.initialization ?? Promise.resolve(); - return previousInitialization.catch(() => undefined).then(initialize); + const initialization = previousInitialization.catch(() => undefined).then(initialize); + void initialization.catch((error) => { + if (this.initialization === initialization) this.initializationFailed = true; + this.logger.error(error); + }); + return initialization; } private async ensureNotificationDirectories(basePath: string) { @@ -170,18 +180,25 @@ export class NotificationsService { ignoreInitial: true, }); NotificationsService.watcher = watcher; + watcher.on('error', (error) => this.logger.error(error)); for (const event of ['add', 'change', 'unlink'] as const) { watcher.on(event, (path) => { void this.handleNotificationEvent(path, event).catch((e) => this.logger.error(e)); }); } - await new Promise((resolve, reject) => { - watcher.once('error', reject); - watcher.once('ready', () => { - watcher.off('error', reject); - resolve(); + try { + await new Promise((resolve, reject) => { + watcher.once('error', reject); + watcher.once('ready', () => { + watcher.off('error', reject); + resolve(); + }); }); - }); + } catch (error) { + if (NotificationsService.watcher === watcher) NotificationsService.watcher = null; + await watcher.close().catch((closeError) => this.logger.error(closeError)); + throw error; + } return watcher; } From d540c61734a6abad4cb0a72663050a74c6929e48 Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sun, 25 Oct 2026 13:38:58 -0400 Subject: [PATCH 6/8] test(notifications): preserve appliance fault diagnostics --- api/scripts/qa/notification-counter.mjs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/api/scripts/qa/notification-counter.mjs b/api/scripts/qa/notification-counter.mjs index 541365e4b6..f97b717965 100644 --- a/api/scripts/qa/notification-counter.mjs +++ b/api/scripts/qa/notification-counter.mjs @@ -297,6 +297,7 @@ async function adversarial() { const saved = join(notificationPath, `${prefix}_saved_unread`); await rename(unread, saved); await writeFile(unread, 'blocked directory'); + let startupError; try { execFileSync('/etc/rc.d/rc.unraid-api', ['restart'], { timeout: 60000, stdio: 'pipe' }); const failedProcess = apiProcess(); @@ -309,9 +310,17 @@ async function adversarial() { await rename(saved, unread); await verify('recovers startup directory failure without another restart', { unread: 2, archive: 0 }); assert.deepEqual(apiProcess(), failedProcess, 'Process restarted instead of recovering'); + } catch (error) { + startupError = error; + throw error; } finally { try { await access(saved); await rm(unread, { recursive: true, force: true }); await rename(saved, unread); } - catch (error) { if (error.code !== 'ENOENT') throw error; } + catch (error) { + if (error.code !== 'ENOENT') { + report.startupCleanupErrorType = error.name; + if (!startupError) throw error; + } + } } await mutation('archiveNotification', persistent); await verify('mutation after initialization recovery', { unread: 1, archive: 1 }); @@ -326,7 +335,7 @@ async function adversarial() { try { await writeFile(limitPath, '1\n'); await mkdir(faultDirectory); - await waitUntil(async () => (await readFile(logPath, 'utf8')).slice(oldLogLength).includes('ENOSPC'), + await waitUntil(async () => (await readFile(logPath)).subarray(oldLogLength).toString('utf8').includes('ENOSPC'), 'actual inotify resource exhaustion'); } finally { await writeFile(limitPath, originalLimit); From 6f9a94d0d46e96a7e2b8e15107f008e8cbff043a Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sun, 25 Oct 2026 16:32:05 -0400 Subject: [PATCH 7/8] test(notifications): measure burst read amplification --- api/scripts/qa/README.md | 37 +++ .../notifications/notifications.load.spec.ts | 222 ++++++++++++++++++ 2 files changed, 259 insertions(+) create mode 100644 api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts diff --git a/api/scripts/qa/README.md b/api/scripts/qa/README.md index ec56699d84..1f24cbad2b 100644 --- a/api/scripts/qa/README.md +++ b/api/scripts/qa/README.md @@ -1,5 +1,42 @@ # Notification counter appliance test +## Local burst and read-cost tests + +From the repository root, run: + +```sh +pnpm --filter ./api test src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts +CHOKIDAR_USEPOLLING=true pnpm --filter ./api test src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts +``` + +These tests use real temporary files, Chokidar events, notification readers, +and PubSub subscriptions. They cover creation of both native file copies, +content/severity changes, unread deletion revealing archive copies, and +archive-only creation. Fixtures include 1,000 unread plus 1,000 archive history +files and bursts of 24 or 100 notifications. + +Each `notification-load` record reports file-content read attempts, warning/alert +scan count, peak simultaneous scans, and elapsed time. Measurement starts after +hydration and ends after handlers and payload publications finish. Ground-truth +list queries run afterward and do not inflate the recorded event cost. Read +attempts include missing-copy reads; they are not physical disk I/O measurements. +Watcher metadata reads and other processes are outside this measurement. + +Assertions check final counters against full lists, final warning payloads, +deduplicated additions, no history recount, no reads of unrelated archived +payloads, and no more than two count reads per event plus the unread payload +scans. Archive-only bursts must trigger no warning/alert scans. + +**These are cost-characterization tests, not a burst-performance approval.** +The current implementation starts a warning/alert scan per unread event. With +1,000 unread history files, a burst of 100 changes produced 110,200 content-read +attempts and 100 scans in a local run. Passing tests preserve notification +correctness and expose that amplification; they do not establish an acceptable +server latency or a constant read budget per burst. Elapsed times are diagnostic, +not machine-dependent CI limits. Appliance acceptance remains a separate step. + +## Appliance setup + `notification-counter.mjs` tests the installed API against Unraid's native `/usr/local/emhttp/webGui/scripts/notify`. Run it inside a disposable, managed QA guest after installing the production-shaped plugin for the selected commit. diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts new file mode 100644 index 0000000000..a3eeab25af --- /dev/null +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts @@ -0,0 +1,222 @@ +import { ConfigService } from '@nestjs/config'; +import { copyFile, mkdir, mkdtemp, rm, unlink, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { basename, join } from 'path'; + +import type { FSWatcher } from 'chokidar'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { pubsub, PUBSUB_CHANNEL } from '@app/core/pubsub.js'; +import { + Notification, + NotificationImportance, + NotificationType, +} from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; + +const { UNREAD, ARCHIVE } = NotificationType; + +describe.sequential('Notification burst read costs', () => { + let basePath: string; + let service: NotificationsService; + let watcher: FSWatcher; + const subscriptions: number[] = []; + const handlers: Promise[] = []; + const publications: Promise[] = []; + + afterEach(async () => { + await watcher?.close(); + Reflect.set(NotificationsService, 'watcher', null); + await Promise.allSettled(handlers.splice(0)); + await Promise.allSettled(publications.splice(0)); + subscriptions.splice(0).forEach((id) => pubsub.unsubscribe(id)); + vi.restoreAllMocks(); + if (basePath) await rm(basePath, { recursive: true, force: true }); + }); + + for (const [historySize, burstSize] of [ + [0, 24], + [1000, 24], + [1000, 100], + ]) { + it.each(['create', 'change', 'delete', 'archive-only'] as const)( + `measures ${burstSize} %s events with ${historySize} unread and archived history files`, + async (operation) => { + basePath = await mkdtemp(join(tmpdir(), 'notification-load-')); + const path = (type: NotificationType, id: string) => + join(basePath, type.toLowerCase(), id); + const write = (type: NotificationType, id: string, importance = 'normal') => + writeFile( + path(type, id), + `timestamp=1700000000\nevent=${id}\nsubject=Load\ndescription=Load\nimportance=${importance}\n` + ); + await Promise.all([mkdir(path(UNREAD, '')), mkdir(path(ARCHIVE, ''))]); + for (const type of [UNREAD, ARCHIVE]) { + await Promise.all( + Array.from({ length: historySize }, (_, index) => + write(type, `${type.toLowerCase()}-history-${index}.notify`) + ) + ); + } + const ids = Array.from({ length: burstSize }, (_, index) => `burst-${index}.notify`); + if (operation === 'change' || operation === 'delete') { + await Promise.all( + ids.map(async (id) => { + await write(UNREAD, id); + await write(ARCHIVE, id, 'alert'); + }) + ); + } + + service = new NotificationsService( + new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }) + ); + await Reflect.get(service, 'initialization'); + watcher = Reflect.get(NotificationsService, 'watcher'); + const reads: string[] = []; + const events: { kind: string; path: string }[] = []; + const failures: unknown[] = []; + const snapshots: Notification[][] = []; + const additions: Notification[] = []; + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, + (payload: { notificationsWarningsAndAlerts: Notification[] }) => + snapshots.push(payload.notificationsWarningsAndAlerts) + ), + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_ADDED, + (payload: { notificationAdded: Notification }) => + additions.push(payload.notificationAdded) + ) + ); + const load = Reflect.get(service, 'loadNotificationFile').bind(service); + Reflect.set(service, 'loadNotificationFile', (file: string, type: NotificationType) => { + reads.push(file); + return load(file, type); + }); + const handle = Reflect.get(service, 'handleNotificationEvent').bind(service); + Reflect.set(service, 'handleNotificationEvent', (file: string, kind: string) => { + events.push({ kind, path: file }); + const pending = handle(file, kind).catch((error: unknown) => failures.push(error)); + handlers.push(pending); + return pending; + }); + const publish = Reflect.get(service, 'publishWarningsAndAlerts').bind(service); + Reflect.set(service, 'publishWarningsAndAlerts', () => { + const pending = publish(); + publications.push(pending); + return pending; + }); + let activeScans = 0; + let peakActiveScans = 0; + const readWarnings = service.getWarningsAndAlerts.bind(service); + const scans = vi.spyOn(service, 'getWarningsAndAlerts').mockImplementation(async () => { + activeScans++; + peakActiveScans = Math.max(peakActiveScans, activeScans); + try { + return await readWarnings(); + } finally { + activeScans--; + } + }); + const recount = vi.spyOn(service, 'recalculateOverview'); + const started = performance.now(); + await Promise.all( + ids.map(async (id) => { + switch (operation) { + case 'create': + await write(UNREAD, id, 'warning'); + await copyFile(path(UNREAD, id), path(ARCHIVE, id)); + break; + case 'change': + await write(UNREAD, id, 'warning'); + break; + case 'delete': + await unlink(path(UNREAD, id)); + break; + case 'archive-only': + await write(ARCHIVE, id, 'warning'); + break; + } + }) + ); + const expectedEvents = burstSize * (operation === 'create' ? 2 : 1); + await vi.waitFor(() => expect(events).toHaveLength(expectedEvents), { + timeout: 15000, + }); + await watcher.close(); + await Promise.all(handlers); + await Promise.all(publications); + const elapsedMs = performance.now() - started; + const measuredReads = [...reads]; + expect(failures).toEqual([]); + expect(recount).not.toHaveBeenCalled(); + expect( + measuredReads.filter((file) => basename(file).startsWith('archive-history-')) + ).toEqual([]); + const unreadEvents = events.filter((event) => event.path.includes('/unread/')); + expect(scans.mock.calls.length).toBeLessThanOrEqual(unreadEvents.length); + const maximumUnread = historySize + (operation === 'archive-only' ? 0 : burstSize); + expect(measuredReads.length).toBeLessThanOrEqual( + 2 * expectedEvents + maximumUnread * scans.mock.calls.length + ); + const unreadTotal = + historySize + + (operation === 'delete' || operation === 'archive-only' ? 0 : burstSize); + const archiveTotal = + historySize + + (operation === 'delete' || operation === 'archive-only' ? burstSize : 0); + expect(service.getOverview()).toMatchObject({ + unread: { + total: unreadTotal, + warning: operation === 'create' || operation === 'change' ? burstSize : 0, + }, + archive: { total: archiveTotal, alert: operation === 'delete' ? burstSize : 0 }, + }); + if (unreadEvents.length) { + expect(snapshots.length).toBeGreaterThan(0); + const latest = snapshots.at(-1) ?? []; + expect(latest).toHaveLength(operation === 'delete' ? 0 : Math.min(burstSize, 50)); + for (const notification of latest) { + expect(ids).toContain(notification.id); + expect(notification.importance).toBe(NotificationImportance.WARNING); + } + } else { + expect(scans).not.toHaveBeenCalled(); + expect(snapshots).toEqual([]); + } + expect(additions).toHaveLength(operation === 'create' ? burstSize : 0); + expect(new Set(additions.map((notification) => notification.id)).size).toBe( + additions.length + ); + + // Ground-truth list reads are deliberately outside the measured event workload. + for (const type of [UNREAD, ARCHIVE]) { + const list = await service.getNotifications({ type, offset: 0, limit: 10000 }); + const counts = service.getOverview()[type.toLowerCase()]; + expect(counts.total).toBe(list.length); + for (const importance of Object.values(NotificationImportance)) { + expect(counts[importance.toLowerCase()]).toBe( + list.filter((notification) => notification.importance === importance).length + ); + } + } + console.info( + 'notification-load', + JSON.stringify({ + operation, + historySize, + burstSize, + events: expectedEvents, + fileReadAttempts: measuredReads.length, + warningScans: scans.mock.calls.length, + peakActiveScans, + elapsedMs: Math.round(elapsedMs), + }) + ); + }, + 60000 + ); + } +}); From acd1994b4ec8e48ac6f1a19dd006c5f7932cff0e Mon Sep 17 00:00:00 2001 From: Eli Bosley Date: Sun, 25 Oct 2026 16:43:24 -0400 Subject: [PATCH 8/8] perf(notifications): combine overlapping warning refreshes --- api/scripts/qa/README.md | 24 ++- .../notifications.broadcast.spec.ts | 170 ++++++++++++++++++ .../notifications/notifications.load.spec.ts | 2 + .../notifications.overview.spec.ts | 5 +- .../notifications.service.spec.ts | 1 + .../notifications/notifications.service.ts | 40 +++-- 6 files changed, 220 insertions(+), 22 deletions(-) create mode 100644 api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts diff --git a/api/scripts/qa/README.md b/api/scripts/qa/README.md index 1f24cbad2b..1abaf72478 100644 --- a/api/scripts/qa/README.md +++ b/api/scripts/qa/README.md @@ -25,15 +25,21 @@ Watcher metadata reads and other processes are outside this measurement. Assertions check final counters against full lists, final warning payloads, deduplicated additions, no history recount, no reads of unrelated archived payloads, and no more than two count reads per event plus the unread payload -scans. Archive-only bursts must trigger no warning/alert scans. - -**These are cost-characterization tests, not a burst-performance approval.** -The current implementation starts a warning/alert scan per unread event. With -1,000 unread history files, a burst of 100 changes produced 110,200 content-read -attempts and 100 scans in a local run. Passing tests preserve notification -correctness and expose that amplification; they do not establish an acceptable -server latency or a constant read budget per burst. Elapsed times are diagnostic, -not machine-dependent CI limits. Appliance acceptance remains a separate step. +scans. They require at most one active payload scan and at most four scans for +each large-history test burst. Archive-only bursts must trigger no warning/alert +scans. This limits the fixture's total read budget rather than only its counter +reads. Separate broadcast tests hold reads/publication open to prove that +overlapping requests combine, a mid-scan request produces a fresh final payload, +and read/publication failures do not block later requests. + +Before refresh scheduling, 100 changes with 1,000 unread history files caused +110,200 content-read attempts and 100 overlapping scans. With scheduling, the +same native-watcher workload used 1,300 attempts and one scan in a local run; +polling used 2,400 attempts and two sequential scans. Incoming events during a +scan can require a follow-up scan; there is no fixed scan-count guarantee for an +unbounded stream of events. Elapsed times remain diagnostic, not machine-dependent +CI limits. These local tests do not establish appliance latency or physical disk +I/O performance. Appliance acceptance remains a separate step. ## Appliance setup diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts new file mode 100644 index 0000000000..80a24987db --- /dev/null +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts @@ -0,0 +1,170 @@ +import { ConfigService } from '@nestjs/config'; +import { mkdir, mkdtemp, rm, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import type { FSWatcher } from 'chokidar'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { pubsub, PUBSUB_CHANNEL } from '@app/core/pubsub.js'; +import { + Notification, + NotificationImportance, +} from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; + +describe.sequential('Warning and alert refresh scheduling', () => { + let basePath: string; + let service: NotificationsService; + let subscription: number; + let snapshots: Notification[][]; + const pending: Promise[] = []; + const gates: (() => void)[] = []; + + const write = (importance: string) => + writeFile( + join(basePath, 'unread', 'current.notify'), + `timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=${importance}\n` + ); + function refresh(): Promise { + const result = Reflect.get(service, 'publishWarningsAndAlerts').call(service); + pending.push(result); + return result; + } + function gate() { + const result = Promise.withResolvers(); + gates.push(result.resolve); + return result; + } + + beforeEach(async () => { + basePath = await mkdtemp(join(tmpdir(), 'notification-broadcast-')); + await mkdir(join(basePath, 'unread')); + await mkdir(join(basePath, 'archive')); + await write('warning'); + service = new NotificationsService( + new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }) + ); + await Reflect.get(service, 'initialization'); + const watcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + await watcher.close(); + Reflect.set(NotificationsService, 'watcher', null); + snapshots = []; + subscription = await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, + (payload: { notificationsWarningsAndAlerts: Notification[] }) => + snapshots.push(payload.notificationsWarningsAndAlerts) + ); + }); + + afterEach(async () => { + gates.splice(0).forEach((release) => release()); + await Promise.allSettled(pending.splice(0)); + pubsub.unsubscribe(subscription); + vi.restoreAllMocks(); + await rm(basePath, { recursive: true, force: true }); + }); + + it('combines simultaneous refresh requests into one real payload read', async () => { + const scans = vi.spyOn(service, 'getWarningsAndAlerts'); + await Promise.all(Array.from({ length: 100 }, refresh)); + expect(scans).toHaveBeenCalledTimes(1); + expect(snapshots).toHaveLength(1); + expect(snapshots[0][0].importance).toBe(NotificationImportance.WARNING); + }); + + it('serializes a mid-read burst and publishes the latest disk state afterward', async () => { + const captured = gate(); + const release = gate(); + const read = service.getWarningsAndAlerts.bind(service); + let active = 0; + let peak = 0; + const scans = vi.spyOn(service, 'getWarningsAndAlerts').mockImplementation(async () => { + active++; + peak = Math.max(peak, active); + try { + const snapshot = await read(); + if (!snapshots.length) { + captured.resolve(); + await release.promise; + } + return snapshot; + } finally { + active--; + } + }); + refresh(); + await captured.promise; + await write('alert'); + Array.from({ length: 100 }, refresh); + release.resolve(); + await Promise.all(pending); + expect(peak).toBe(1); + expect(scans).toHaveBeenCalledTimes(2); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.ALERT); + }); + + it('keeps a request received during a failed scan and accepts later requests', async () => { + const captured = gate(); + const release = gate(); + const scans = vi.spyOn(service, 'getWarningsAndAlerts').mockImplementationOnce(async () => { + captured.resolve(); + await release.promise; + throw new Error('temporary notification read failure'); + }); + const logger = Reflect.get(service, 'logger'); + const errors = vi.spyOn(logger, 'error'); + refresh(); + await captured.promise; + await write('alert'); + refresh(); + release.resolve(); + await Promise.all(pending); + expect(scans).toHaveBeenCalledTimes(2); + expect(errors).toHaveBeenCalled(); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.ALERT); + await write('warning'); + await refresh(); + expect(scans).toHaveBeenCalledTimes(3); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.WARNING); + }); + + it('does not lose refresh requests received while publication is pending', async () => { + const captured = gate(); + const release = gate(); + const publish = pubsub.publish.bind(pubsub); + vi.spyOn(pubsub, 'publish').mockImplementationOnce(async (channel, payload) => { + captured.resolve(); + await release.promise; + return publish(channel, payload); + }); + const scans = vi.spyOn(service, 'getWarningsAndAlerts'); + refresh(); + await captured.promise; + await write('alert'); + Array.from({ length: 100 }, refresh); + release.resolve(); + await Promise.all(pending); + expect(scans).toHaveBeenCalledTimes(2); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.ALERT); + }); + + it.each(['read', 'publication'] as const)( + 'releases the scheduler after a %s failure with no pending request', + async (failure) => { + if (failure === 'read') { + vi.spyOn(service, 'getWarningsAndAlerts').mockRejectedValueOnce( + new Error('read failed') + ); + } else { + vi.spyOn(pubsub, 'publish').mockRejectedValueOnce(new Error('publication failed')); + } + await refresh(); + expect(snapshots).toEqual([]); + await write('alert'); + await refresh(); + expect(snapshots).toHaveLength(1); + expect(snapshots[0][0].importance).toBe(NotificationImportance.ALERT); + } + ); +}); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts index a3eeab25af..89c589c349 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts @@ -157,6 +157,8 @@ describe.sequential('Notification burst read costs', () => { ).toEqual([]); const unreadEvents = events.filter((event) => event.path.includes('/unread/')); expect(scans.mock.calls.length).toBeLessThanOrEqual(unreadEvents.length); + expect(peakActiveScans).toBeLessThanOrEqual(1); + if (historySize) expect(scans.mock.calls.length).toBeLessThanOrEqual(4); const maximumUnread = historySize + (operation === 'archive-only' ? 0 : burstSize); expect(measuredReads.length).toBeLessThanOrEqual( 2 * expectedEvents + maximumUnread * scans.mock.calls.length diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts index 61e081bc66..71afe29bd5 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts @@ -125,7 +125,10 @@ describe.sequential('Notification filesystem and subscription integration', () = const activeWatcher: FSWatcher | null = Reflect.get(NotificationsService, 'watcher'); await activeWatcher?.close(); Reflect.set(NotificationsService, 'watcher', null); - if (service) await Reflect.get(service, 'overviewUpdate'); + if (service) { + await Reflect.get(service, 'overviewUpdate'); + await Reflect.get(service, 'warningsUpdate'); + } subscriptions.forEach((subscription) => pubsub.unsubscribe(subscription)); vi.restoreAllMocks(); vi.unstubAllGlobals(); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts index 8daeaadb67..cc71feb67e 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts @@ -100,6 +100,7 @@ describe.sequential('NotificationsService', () => { // make sure each test is isolated (as much as possible) afterEach(async () => { await service.deleteAllNotifications(); + await Reflect.get(service, 'warningsUpdate'); expect(service.getOverview()).toEqual(zeroOverview()); }); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts index ec6a30895b..35a3149d09 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts @@ -48,6 +48,8 @@ export class NotificationsService { private initialization: Promise | null = null; private initializationFailed = false; private overviewUpdate: Promise = Promise.resolve(); + private warningsUpdate: Promise | null = null; + private warningsUpdateRequested = false; private notificationFiles = new Map(); private isHydratingOverview = false; private pendingNotificationEvents = new Map(); @@ -303,18 +305,32 @@ export class NotificationsService { }); } - private async publishWarningsAndAlerts() { - try { - const warningsAndAlerts = await this.getWarningsAndAlerts(); - await pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, { - notificationsWarningsAndAlerts: warningsAndAlerts, - }); - } catch (error) { - this.logger.error( - '[publishWarningsAndAlerts] Failed to broadcast warnings and alerts snapshot', - error as Error - ); - } + private publishWarningsAndAlerts(): Promise { + this.warningsUpdateRequested = true; + if (this.warningsUpdate) return this.warningsUpdate; + + this.warningsUpdate = Promise.resolve().then(async () => { + try { + while (this.warningsUpdateRequested) { + await this.overviewUpdate.catch(() => undefined); + this.warningsUpdateRequested = false; + try { + const warningsAndAlerts = await this.getWarningsAndAlerts(); + await pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, { + notificationsWarningsAndAlerts: warningsAndAlerts, + }); + } catch (error) { + this.logger.error( + '[publishWarningsAndAlerts] Failed to broadcast warnings and alerts snapshot', + error as Error + ); + } + } + } finally { + this.warningsUpdate = null; + } + }); + return this.warningsUpdate; } private increment(importance: NotificationImportance, collector: NotificationCounts) {