feat(reactive-rpc): 🎸 remove old blocks after some period of time

Author: streamich

src/server/services/blocks/BlocksServices.ts

@@ -3,12 +3,15 @@ import {StorePatch} from './types';
 import {RpcError, RpcErrorCodes} from '../../../reactive-rpc/common/rpc/caller';
 import type {Services} from '../Services';
 
+const BLOCK_TTL = 1000 * 60 * 20; // 20 minutes
+
 export class BlocksServices {
   protected readonly store = new MemoryStore();
 
   constructor(protected readonly services: Services) {}
 
   public async create(id: string, patches: StorePatch[]) {
+    this.maybeGc();
     const {store} = this;
     const {block} = await store.create(id, patches);
     const data = {
@@ -45,6 +48,7 @@ export class BlocksServices {
   }
 
   public async edit(id: string, patches: any[]) {
+    this.maybeGc();
     if (!Array.isArray(patches)) throw RpcError.validation('patches must be an array');
     if (!patches.length) throw RpcError.validation('patches must not be empty');
     const seq = patches[0].seq;
@@ -69,4 +73,18 @@ export class BlocksServices {
   public stats() {
     return this.store.stats();
   }
+
+  private maybeGc(): void {
+    if (Math.random() < 0.05)
+      this.gc().catch((error) => {
+        // tslint:disable-next-line:no-console
+        console.error('Error running gc', error);
+      });
+  }
+
+  private async gc(): Promise<void> {
+    const ts = Date.now() - BLOCK_TTL;
+    const {store} = this;
+    await store.removeOlderThan(ts);
+  }
 }

src/server/services/blocks/MemoryStore.ts

@@ -72,6 +72,10 @@ export class MemoryStore implements types.Store {
 
   public async remove(id: string): Promise<void> {
     await new Promise((resolve) => setImmediate(resolve));
+    this.removeSync(id);
+  }
+
+  private removeSync(id: string): void {
     this.blocks.delete(id);
     this.patches.delete(id);
   }
@@ -82,4 +86,9 @@ export class MemoryStore implements types.Store {
       patches: [...this.patches.values()].reduce((acc, v) => acc + v.length, 0),
     };
   }
+
+  public async removeOlderThan(ts: number): Promise<void> {
+    await new Promise((resolve) => setImmediate(resolve));
+    for (const [id, block] of this.blocks) if (block.created < ts) this.removeSync(id);
+  }
 }