log-client: check missing logs on server after flushing (#257)

* log-client: check missing logs after flushing

* log-client: check for missing logs after flushing

* changeset

Author: QuentinRoy

.changeset/stale-terms-stare.md

@@ -0,0 +1,5 @@
+---
+"@lightmill/log-client": minor
+---
+
+`Logger#flush` now checks for missing logs on the server and fails if any are missing that were added prior to the flush call. This is considered a minor change, as it primarily surfaces existing issues (such as communication errors) earlier.

packages/log-client/__mocks__/mock-server.ts

@@ -305,7 +305,27 @@ export class MockServer {
 
       '/runs/{id}': {
         get: async () => {
-          throw new Error('Not implemented: GET /runs/{id}');
+          return {
+            status: 200,
+            body: {
+              data: {
+                id: 'default-run-id',
+                type: 'runs' as const,
+                attributes: {
+                  name: 'Default Run',
+                  status: 'idle',
+                  lastLogNumber: 0,
+                  missingLogNumbers: [],
+                },
+                relationships: {
+                  experiment: {
+                    data: { id: 'default-experiment-id', type: 'experiments' },
+                  },
+                  lastLogs: { data: [] },
+                },
+              },
+            },
+          };
         },
         patch: async ({ params, body }) => {
           const run = this.#runs.get(params.id as string);

packages/log-client/__tests__/logger.test.ts

@@ -283,26 +283,14 @@ describe('LogClient#flush', () => {
         body: { data: { id: `log-id-${reqManager.size() + 1}`, type: 'logs' } },
       });
     });
-    logger.addLog({
-      type: 'mock-log',
-      val: 1,
-      date: new Date('2021-06-03T02:00:00.000Z'),
-    });
-    logger.addLog({
-      type: 'mock-log',
-      val: 2,
-      date: new Date('2021-06-03T03:00:00.000Z'),
-    });
+    logger.addLog({ type: 'mock-log', val: 1 });
+    logger.addLog({ type: 'mock-log', val: 2 });
     let resolved = false;
     let flushPromise = logger.flush().then((result) => {
       resolved = true;
       return result;
     });
-    logger.addLog({
-      type: 'mock-log',
-      val: 3,
-      date: new Date('2021-06-03T04:00:00.000Z'),
-    });
+    logger.addLog({ type: 'mock-log', val: 3 });
     await reqManager.waitForRequests(2);
     expect(resolved).toBe(false);
     reqManager.resolveNextRequest();
@@ -328,38 +316,53 @@ describe('LogClient#flush', () => {
         body: { data: { id: `log-id-${defManager.size() + 1}`, type: 'logs' } },
       });
     });
-    logger.addLog({
-      type: 'mock-log',
-      val: 1,
-      date: new Date('2021-06-03T02:00:00.000Z'),
-    });
-    logger.addLog({
-      type: 'mock-log',
-      val: 2,
-      date: new Date('2021-06-03T03:00:00.000Z'),
+    let oldGetRun = server.handlers['/runs/{id}'].get.getMockImplementation()!;
+    server.handlers['/runs/{id}'].get.mockImplementation(async (...args) => {
+      let result = await oldGetRun(...args);
+      if (result.status !== 200) return result;
+      result.body.data.attributes = {
+        ...result.body.data.attributes,
+        // Should be ignored by first flush call.
+        missingLogNumbers: [3],
+      };
+      return result;
     });
+    logger.addLog({ type: 'mock-log', val: 1 });
+    logger.addLog({ type: 'mock-log', val: 2 });
     let flushPromise = logger.flush();
-    logger
-      .addLog({
-        type: 'mock-log',
-        val: 'fail',
-        date: new Date('2021-06-03T04:00:00.000Z'),
-      })
-      .catch(() => {
-        // Prevent vitest from catching the error and complaining about it.
-      });
-    logger.addLog({
-      type: 'mock-log',
-      val: 4,
-      date: new Date('2021-06-03T03:00:00.000Z'),
+    logger.addLog({ type: 'mock-log', val: 'fail' }).catch(() => {
+      // Prevent vitest from catching the error and complaining about it.
     });
+    logger.addLog({ type: 'mock-log', val: 4 });
     await defManager.waitForRequests(3);
     defManager.resolveAllRequests();
     await expect(flushPromise).resolves.toBeUndefined();
     await expect(logger.flush()).rejects.toThrowErrorMatchingInlineSnapshot(
       `[AddLogError: FORBIDDEN]`,
     );
   });
+
+  it('fails if there are still missing logs on the server', async ({
+    logger,
+    server,
+  }) => {
+    let oldGetRun = server.handlers['/runs/{id}'].get.getMockImplementation()!;
+    server.handlers['/runs/{id}'].get.mockImplementation(async (...args) => {
+      let result = await oldGetRun(...args);
+      if (result.status !== 200) return result;
+      result.body.data.attributes = {
+        ...result.body.data.attributes,
+        // Should be ignored by first flush call.
+        missingLogNumbers: [1],
+      };
+      return result;
+    });
+    logger.addLog({ type: 'mock-log', val: 1 });
+    logger.addLog({ type: 'mock-log', val: 2 });
+    await expect(logger.flush()).rejects.toThrowErrorMatchingInlineSnapshot(
+      `[FlushError: There are missing logs on server after flushing. Missing logs: 1]`,
+    );
+  });
 });
 
 describe('LogClient#completeRun', () => {

packages/log-client/src/logger.ts

@@ -135,6 +135,16 @@ export class LightmillLogger<
         },
       });
     });
+    const missingLogs = await this.#fetchMissingLogs();
+    if (missingLogs.some((l) => l <= lastLogNumber)) {
+      // If we have missing logs that are before or at the last log number,
+      // then we know that there are missing logs on the server that we
+      // do not have in our pending logs, which means we lost some logs
+      // in the process.
+      throw new FlushError(
+        `There are missing logs on server after flushing. Missing logs: ${missingLogs.join(', ')}`,
+      );
+    }
   }
 
   #isTherePendingLogsBefore(logNumber: number): boolean {
@@ -145,6 +155,18 @@ export class LightmillLogger<
     return first != null && first <= logNumber;
   }
 
+  async #fetchMissingLogs() {
+    const response = await this.#fetchClient.GET('/runs/{id}', {
+      credentials: 'include',
+      params: { path: { id: this.#runId } },
+      headers: { 'content-type': apiMediaType },
+    });
+    if (response.error != null) {
+      throw new RequestError(response);
+    }
+    return response.data.data.attributes.missingLogNumbers;
+  }
+
   async completeRun() {
     await this.#endRun('completed');
   }
@@ -196,3 +218,10 @@ class AddLogError extends Error {
     this.logNumber = logNumber;
   }
 }
+
+class FlushError extends Error {
+  name = 'FlushError' as const;
+  constructor(message: string) {
+    super(message);
+  }
+}