CommunitySolidServer
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
@@ -31,6 +31,10 @@ The following changes pertain to the imports in the default configs:
 
 The following changes are relevant for v7 custom configs that replaced certain features.
 
+- `KeyValueChannelStorage` now sweeps expired notification channels every 60 minutes by default,
+  with up to 15% jitter between instances.
+  Custom configurations can set the interval in minutes (`0` disables the sweep) and the jitter fraction,
+  and should register the storage with the `Finalizer` so its timer is cleared during shutdown.
 - Due to extracting the core handlers as an external library,
   the CSS had to adapt some of them resulting in new class names.
   The following renames have happened, meaning that if you used or extended a component of one of the following types,

diff --git a/config/http/notifications/base/storage.json b/config/http/notifications/base/storage.json
@@ -1,5 +1,8 @@
 {
-  "@context": "https://linkedsoftwaredependencies.org/bundles/npm/@solid/community-server/^8.0.0/components/context.jsonld",
+  "@context": [
+    "https://linkedsoftwaredependencies.org/bundles/npm/@solid/community-server/^8.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/asynchronous-handlers/^1.0.0/components/context.jsonld"
+  ],
   "@graph": [
     {
       "comment": "Storage to be used to keep track of subscriptions.",
@@ -11,6 +14,17 @@
         "relativePath": "/notifications/",
         "source": { "@id": "urn:solid-server:default:KeyValueStorage" }
       }
+    },
+    {
+      "comment": "Makes sure the notification channel sweep timer is stopped when the application stops.",
+      "@id": "urn:solid-server:default:Finalizer",
+      "@type": "ParallelHandler",
+      "handlers": [
+        {
+          "@type": "FinalizableHandler",
+          "finalizable": { "@id": "urn:solid-server:default:SubscriptionStorage" }
+        }
+      ]
     }
   ]
 }
diff --git a/src/server/notifications/KeyValueChannelStorage.ts b/src/server/notifications/KeyValueChannelStorage.ts
@@ -1,8 +1,10 @@
 import { getLoggerFor } from 'global-logger-factory';
 import type { ResourceIdentifier } from '../../http/representation/ResourceIdentifier';
+import type { Finalizable } from '../../init/final/Finalizable';
 import type { KeyValueStorage } from '../../storage/keyvalue/KeyValueStorage';
 import { InternalServerError } from '../../util/errors/InternalServerError';
 import type { ReadWriteLocker } from '../../util/locking/ReadWriteLocker';
+import { setSafeInterval } from '../../util/TimerUtil';
 import type { NotificationChannel } from './NotificationChannel';
 import type { NotificationChannelStorage } from './NotificationChannelStorage';
 
@@ -13,16 +15,46 @@ type StorageValue = string | string[] | NotificationChannel;
  * Encodes IDs/topics before storing them in the KeyValueStorage.
  *
  * Uses a {@link ReadWriteLocker} to prevent internal race conditions.
+ *
+ * Expired channels are deleted when they are requested through `get`.
+ * A timer additionally deletes all expired channels periodically,
+ * since channels that are never requested again would otherwise remain in the storage.
  */
-export class KeyValueChannelStorage implements NotificationChannelStorage {
+export class KeyValueChannelStorage implements NotificationChannelStorage, Finalizable {
   protected logger = getLoggerFor(this);
 
   private readonly storage: KeyValueStorage<string, StorageValue>;
   private readonly locker: ReadWriteLocker;
+  private readonly timer?: NodeJS.Timeout;
+  private activeSweep?: Promise<void>;
 
-  public constructor(storage: KeyValueStorage<string, StorageValue>, locker: ReadWriteLocker) {
+  /**
+   * @param storage - Where to store the channels.
+   * @param locker - Used to prevent internal race conditions.
+   * @param sweepInterval - How often the expired channels need to be deleted, in minutes. `0` disables the sweep.
+   * @param jitter - Maximum random fraction of `sweepInterval` that is added to the interval,
+   *                 so multiple instances do not all sweep at the same time.
+   */
+  public constructor(
+    storage: KeyValueStorage<string, StorageValue>,
+    locker: ReadWriteLocker,
+    sweepInterval = 60,
+    jitter = 0.15,
+  ) {
     this.storage = storage;
     this.locker = locker;
+
+    if (sweepInterval > 0) {
+      const period = sweepInterval * 60 * 1000;
+      const jitterMs = Math.floor(Math.random() * period * jitter);
+      this.timer = setSafeInterval(
+        this.logger,
+        'Failed to sweep expired notification channels',
+        this.sweepExpiredChannels.bind(this),
+        period + jitterMs,
+      );
+      this.timer.unref();
+    }
   }
 
   public async get(id: string): Promise<NotificationChannel | undefined> {
@@ -69,6 +101,11 @@ export class KeyValueChannelStorage implements NotificationChannelStorage {
         if (channel.topic !== oldChannel.topic) {
           throw new InternalServerError(`Trying to change the topic of a notification channel ${channel.id}`);
         }
+      } else {
+        // The channel might have been deleted while this update was waiting for its lock.
+        // Adding it again also restores its topic index entry.
+        await this.add(channel);
+        return;
       }
 
       await this.storage.set(encodeURIComponent(channel.id), channel);
@@ -110,11 +147,55 @@ export class KeyValueChannelStorage implements NotificationChannelStorage {
     });
   }
 
+  /**
+   * Runs the expiry sweep as a single-flight operation.
+   */
+  private async sweepExpiredChannels(): Promise<void> {
+    if (!this.activeSweep) {
+      this.activeSweep = this.performSweep().finally((): void => {
+        this.activeSweep = undefined;
+      });
+    }
+    await this.activeSweep;
+  }
+
+  /**
+   * Deletes all channels that have expired.
+   */
+  private async performSweep(): Promise<void> {
+    this.logger.debug('Sweeping expired notification channels.');
+    const expired: string[] = [];
+    let removed = 0;
+    // Not deleting while iterating to prevent iterator issues
+    for await (const [ , value ] of this.storage.entries()) {
+      if (this.isChannel(value) && typeof value.endAt === 'number' && value.endAt < Date.now()) {
+        expired.push(value.id);
+      }
+    }
+    for (const id of expired) {
+      await this.locker.withWriteLock(this.getLockKey(id), async(): Promise<void> => {
+        const channel = await this.storage.get(encodeURIComponent(id));
+        if (channel && this.isChannel(channel) && typeof channel.endAt === 'number' && channel.endAt < Date.now()) {
+          await this.deleteChannel(channel);
+          removed += 1;
+        }
+      });
+    }
+    this.logger.debug(`Finished sweeping expired notification channels, removed ${removed}.`);
+  }
+
   private isChannel(value: StorageValue): value is NotificationChannel {
     return Boolean((value as NotificationChannel).id);
   }
 
   private getLockKey(identifier: ResourceIdentifier | string): ResourceIdentifier {
     return { path: `${typeof identifier === 'string' ? identifier : identifier.path}.notification-storage` };
   }
+
+  public async finalize(): Promise<void> {
+    if (this.timer) {
+      clearInterval(this.timer);
+    }
+    await this.activeSweep;
+  }
 }
diff --git a/test/unit/server/notifications/KeyValueChannelStorage.test.ts b/test/unit/server/notifications/KeyValueChannelStorage.test.ts
@@ -1,14 +1,16 @@
+import { EventEmitter } from 'node:events';
 import { getLoggerFor } from 'global-logger-factory';
 import type { Logger } from 'global-logger-factory';
 import type { ResourceIdentifier } from '../../../../src/http/representation/ResourceIdentifier';
 import { KeyValueChannelStorage } from '../../../../src/server/notifications/KeyValueChannelStorage';
 import type { NotificationChannel } from '../../../../src/server/notifications/NotificationChannel';
 import type { KeyValueStorage } from '../../../../src/storage/keyvalue/KeyValueStorage';
 import type { ReadWriteLocker } from '../../../../src/util/locking/ReadWriteLocker';
+import { flushPromises } from '../../../util/Util';
 import resetAllMocks = jest.resetAllMocks;
 
 jest.mock('global-logger-factory', (): any => {
-  const logger: Logger = { info: jest.fn(), error: jest.fn() } as any;
+  const logger: Logger = { info: jest.fn(), error: jest.fn(), debug: jest.fn() } as any;
   return { getLoggerFor: (): Logger => logger };
 });
 
@@ -42,7 +44,8 @@ describe('A KeyValueChannelStorage', (): void => {
       withReadLock: jest.fn(),
     };
 
-    storage = new KeyValueChannelStorage(internalStorage, locker);
+    // Disable the background sweep as it is tested separately
+    storage = new KeyValueChannelStorage(internalStorage, locker, 0);
   });
 
   describe('#get', (): void => {
@@ -120,14 +123,16 @@ describe('A KeyValueChannelStorage', (): void => {
         .toThrow(`Trying to update ${topic} which is not a NotificationChannel.`);
     });
 
-    it('sets the channel if there was no previous value.', async(): Promise<void> => {
+    it('restores the topic index if the channel was deleted before the update acquired its lock.', async():
+    Promise<void> => {
       const newChannel = {
         ...channel,
         state: '123456',
       };
       await expect(storage.update(newChannel)).resolves.toBeUndefined();
-      expect([ ...internalMap.values() ]).toEqual(expect.arrayContaining([
-        newChannel,
+      expect([ ...internalMap.entries() ]).toEqual(expect.arrayContaining([
+        [ encodedTopic, [ channel.id ]],
+        [ encodedId, newChannel ],
       ]));
     });
   });
@@ -166,4 +171,176 @@ describe('A KeyValueChannelStorage', (): void => {
       expect(logger.error).toHaveBeenCalledTimes(2);
     });
   });
+
+  describe('the background sweep', (): void => {
+    // Disable the actual interval and simply check it was created with the correct parameters.
+    // The registered callback is invoked manually to verify its behaviour.
+    let mockInterval: jest.SpyInstance;
+    let mockClear: jest.SpyInstance;
+    let mockRandom: jest.SpyInstance;
+    // We only need a stub timer with an `unref` function since we never let it fire on its own.
+    let mockTimer: { unref: jest.Mock };
+
+    beforeEach((): void => {
+      mockTimer = { unref: jest.fn() };
+      mockInterval = jest.spyOn(globalThis, 'setInterval')
+        .mockImplementation(jest.fn().mockReturnValue(mockTimer));
+      mockClear = jest.spyOn(globalThis, 'clearInterval').mockImplementation(jest.fn());
+      // Fixed jitter source so the scheduled delay is deterministic.
+      mockRandom = jest.spyOn(globalThis.Math, 'random').mockReturnValue(0.5);
+    });
+
+    afterEach((): void => {
+      mockInterval.mockRestore();
+      mockClear.mockRestore();
+      mockRandom.mockRestore();
+    });
+
+    it('schedules the sweep on the configured interval when jitter is disabled.', (): void => {
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      expect(mockInterval).toHaveBeenCalledTimes(1);
+      expect(mockInterval.mock.calls[0]).toHaveLength(2);
+      expect(mockInterval.mock.calls[0][1]).toBe(60 * 1000);
+    });
+
+    it('uses a default 60 minute interval and jitter when none are configured.', (): void => {
+      storage = new KeyValueChannelStorage(internalStorage, locker);
+      expect(mockInterval).toHaveBeenCalledTimes(1);
+      // Default period 60 min = 3600000 ms, plus default jitter floor(0.5 * 3600000 * 0.15) = 270000.
+      expect(mockInterval.mock.calls[0][1]).toBe((60 * 60 * 1000) + 270000);
+    });
+
+    it('adds a jitter fraction to the scheduled sweep interval.', (): void => {
+      // Math.random is 0.5 and jitter is 0.2, so floor(0.5 * 60000 * 0.2) = 6000 is added.
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0.2);
+      expect(mockInterval).toHaveBeenCalledTimes(1);
+      expect(mockInterval.mock.calls[0][1]).toBe((60 * 1000) + 6000);
+    });
+
+    it('unrefs the timer so it does not keep the event loop alive.', (): void => {
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      expect(mockTimer.unref).toHaveBeenCalledTimes(1);
+    });
+
+    it('does not schedule a sweep when the interval is 0.', (): void => {
+      storage = new KeyValueChannelStorage(internalStorage, locker, 0);
+      expect(mockInterval).toHaveBeenCalledTimes(0);
+    });
+
+    it('removes expired channels but keeps active and endless ones when it fires.', async(): Promise<void> => {
+      const activeChannel: NotificationChannel = {
+        id: 'http://example.com/.notifications/active',
+        topic,
+        type: 'WebSocketChannel2023',
+        endAt: Date.now() + (60 * 1000),
+      };
+      const endlessChannel: NotificationChannel = {
+        id: 'http://example.com/.notifications/endless',
+        topic,
+        type: 'WebSocketChannel2023',
+      };
+      channel.endAt = 0;
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      await storage.add(channel);
+      await storage.add(activeChannel);
+      await storage.add(endlessChannel);
+
+      // Invoke the callback that was registered with the interval.
+      await (mockInterval.mock.calls[0][0] as () => Promise<void>)();
+
+      // The expired channel and its index reference are gone; the others remain.
+      expect(internalMap.has(encodedId)).toBe(false);
+      expect(internalMap.has(encodeURIComponent(activeChannel.id))).toBe(true);
+      expect(internalMap.has(encodeURIComponent(endlessChannel.id))).toBe(true);
+      expect(internalMap.get(encodedTopic)).toEqual([ activeChannel.id, endlessChannel.id ]);
+    });
+
+    it('keeps a channel that is renewed before its sweep lock is acquired.', async(): Promise<void> => {
+      channel.endAt = 0;
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      await storage.add(channel);
+
+      const renewed = { ...channel, endAt: Date.now() + (60 * 1000) };
+      jest.mocked(locker.withWriteLock).mockImplementation(async(
+        rid: ResourceIdentifier,
+        whileLocked: () => unknown,
+      ): Promise<unknown> => {
+        if (rid.path === `${channel.id}.notification-storage`) {
+          internalMap.set(encodedId, renewed);
+        }
+        return whileLocked();
+      });
+
+      await (mockInterval.mock.calls[0][0] as () => Promise<void>)();
+
+      expect(internalMap.get(encodedId)).toEqual(renewed);
+      expect(internalMap.get(encodedTopic)).toEqual([ channel.id ]);
+    });
+
+    it('does not start another sweep while one is active.', async(): Promise<void> => {
+      const sweepGate = new EventEmitter();
+      const holdSweep = new Promise<void>((resolve): void => {
+        sweepGate.once('release', resolve);
+      });
+      const entries = jest.spyOn(internalStorage, 'entries').mockImplementation(async function* ():
+      AsyncIterableIterator<[string, NotificationChannel]> {
+        await holdSweep;
+        yield [ encodedId, channel ];
+      });
+      channel.endAt = 0;
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      await storage.add(channel);
+      const sweep = mockInterval.mock.calls[0][0] as () => Promise<void>;
+
+      const firstSweep = sweep();
+      const secondSweep = sweep();
+      await flushPromises();
+      expect(entries).toHaveBeenCalledTimes(1);
+
+      sweepGate.emit('release');
+      await Promise.all([ firstSweep, secondSweep ]);
+      await sweep();
+      expect(entries).toHaveBeenCalledTimes(2);
+    });
+
+    it('waits for the active sweep on finalize.', async(): Promise<void> => {
+      const sweepGate = new EventEmitter();
+      const holdSweep = new Promise<void>((resolve): void => {
+        sweepGate.once('release', resolve);
+      });
+      jest.spyOn(internalStorage, 'entries').mockImplementation(async function* ():
+      AsyncIterableIterator<[string, NotificationChannel]> {
+        await holdSweep;
+        yield [ encodedId, channel ];
+      });
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      const sweep = (mockInterval.mock.calls[0][0] as () => Promise<void>)();
+      await flushPromises();
+
+      let finalized = false;
+      const finalize = storage.finalize().then((): void => {
+        finalized = true;
+      });
+      await flushPromises();
+      expect(mockClear).toHaveBeenCalledWith(mockTimer);
+      expect(finalized).toBe(false);
+
+      sweepGate.emit('release');
+      await Promise.all([ sweep, finalize ]);
+      expect(finalized).toBe(true);
+    });
+
+    it('clears the timer on finalize.', async(): Promise<void> => {
+      storage = new KeyValueChannelStorage(internalStorage, locker, 1, 0);
+      await expect(storage.finalize()).resolves.toBeUndefined();
+      expect(mockClear).toHaveBeenCalledTimes(1);
+      expect(mockClear).toHaveBeenLastCalledWith(mockTimer);
+    });
+
+    it('does not clear a timer on finalize when the sweep is disabled.', async(): Promise<void> => {
+      storage = new KeyValueChannelStorage(internalStorage, locker, 0);
+      await expect(storage.finalize()).resolves.toBeUndefined();
+      expect(mockClear).toHaveBeenCalledTimes(0);
+    });
+  });
 });