Merge branch 'upload/s3' of https://github.com/metatypedev/metatype into MET-432--Uploading--files-

Author: destifo

typegate/deno.lock

@@ -43,6 +43,10 @@ async function initRemoteUploadUrlStore(
   return { redisClient };
 }
 
+async function deinitRemoteUploadUrlStore(urlStore: RemoteUploadUrlStore) {
+  await urlStore.redisClient.quit();
+}
+
 function serializeToRedisValue<T>(value: T): string {
   return JSON.stringify(value);
 }
@@ -248,8 +252,8 @@ export class SharedArtifactStore extends ArtifactStore {
     );
   }
 
-  override close(): Promise<void> {
+  override async close(): Promise<void> {
     this.#s3.destroy();
-    return Promise.resolve(void null);
+    await deinitRemoteUploadUrlStore(this.#uploadUrls);
   }
 }

typegate/src/typegate/artifacts/shared.ts

@@ -65,26 +65,26 @@ Meta.test(
     const port = metaTest.port;
     const gate = `http://localhost:${port}`;
 
-    await metaTest.should("work after deploying artifact to S3", async () => {
-      const s3 = new S3Client(syncConfig.s3);
-      assertEquals((await listObjects(s3, syncConfig.s3Bucket))?.length, 0);
-
-      const { serialized, typegate: gateResponseAdd } = await tgDeploy(tg, {
-        baseUrl: gate,
-        auth,
-        artifactsConfig: {
-          prismaMigration: {
-            globalAction: {
-              create: true,
-              reset: false,
-            },
-            migrationDir: "prisma-migrations",
+    const { serialized, typegate: gateResponseAdd } = await tgDeploy(tg, {
+      baseUrl: gate,
+      auth,
+      artifactsConfig: {
+        prismaMigration: {
+          globalAction: {
+            create: true,
+            reset: false,
           },
-          dir: cwd,
+          migrationDir: "prisma-migrations",
         },
-        typegraphPath: path.join(cwd, "wasmedge.ts"),
-        secrets: {},
-      });
+        dir: cwd,
+      },
+      typegraphPath: path.join(cwd, "wasmedge.ts"),
+      secrets: {},
+    });
+
+    await metaTest.should("work after deploying artifact to S3", async () => {
+      const s3 = new S3Client(syncConfig.s3);
+      assertEquals((await listObjects(s3, syncConfig.s3Bucket))?.length, 2);
 
       assertExists(serialized, "serialized has a value");
       assertEquals(gateResponseAdd, {
@@ -109,6 +109,41 @@ Meta.test(
         })
         .on(engine);
 
+      await engine.terminate();
+      s3.destroy();
+    });
+
+    await metaTest.should("work with multiple typegate instances", async () => {
+      const s3 = new S3Client(syncConfig.s3);
+
+      // typegraphs are pushed to s3 whenever pushed to a typegate
+      assertEquals((await listObjects(s3, syncConfig.s3Bucket))?.length, 3);
+
+      const engine = await metaTest.engineFromDeployed(serialized);
+
+      await gql`
+      query {
+        test_wasi_ts(a: 11, b: 12)
+      }
+    `
+        .expectData({
+          test_wasi_ts: 23,
+        })
+        .on(engine);
+
+      // second engine on the other typegate instance
+      const engine2 = await metaTest.engineFromDeployed(serialized);
+
+      await gql`
+        query {
+          test_wasi_ts(a: 15, b: 2)
+        }
+      `
+        .expectData({
+          test_wasi_ts: 17,
+        })
+        .on(engine2);
+
       const { typegate: gateResponseRem } = await tgRemove(tg, {
         baseUrl: gate,
         auth,
@@ -117,63 +152,8 @@ Meta.test(
       assertEquals(gateResponseRem, { data: { removeTypegraphs: true } });
 
       await engine.terminate();
+      await engine2.terminate();
       s3.destroy();
     });
   },
 );
-
-// Meta.test(
-//   {
-//     name: "WasmEdge Runtime typescript SDK: Multiple typegate instances",
-//     syncConfig,
-//     async setup() {
-//       await cleanUp();
-//     },
-//     async teardown() {
-//       await cleanUp();
-//     },
-//     port: true,
-//     multipleTypegates: 3,
-//   },
-//   async (metaTest) => {
-//     const port = metaTest.port;
-//     const gate = `http://localhost:${port}`;
-//     await metaTest.should("work with multiple typegate instances", async () => {
-//       const s3 = new S3Client(syncConfig.s3);
-//       assertEquals((await listObjects(s3, syncConfig.s3Bucket))?.length, 0);
-
-//       const { serialized, typegate: _gateResponseAdd } = await tgDeploy(tg, {
-//         baseUrl: gate,
-//         auth,
-//         artifactsConfig: {
-//           prismaMigration: {
-//             globalAction: {
-//               create: true,
-//               reset: false,
-//             },
-//             migrationDir: "prisma-migrations",
-//           },
-//           dir: cwd,
-//         },
-//         typegraphPath: path.join(cwd, "wasmedge.ts"),
-//         secrets: {},
-//       });
-
-//       assertEquals((await listObjects(s3, syncConfig.s3Bucket))?.length, 1);
-
-//       const engine = await metaTest.engineFromDeployed(serialized);
-
-//       await gql`
-//       query {
-//         test_wasi_ts(a: 11, b: 2)
-//       }
-//     `
-//         .expectData({
-//           test_wasi_ts: 13,
-//         })
-//         .on(engine);
-//       await engine.terminate();
-//       s3.destroy();
-//     });
-//   },
-// );

typegate/tests/runtimes/wasmedge/wasmedge_sync_test.ts

@@ -33,12 +33,33 @@ export enum SDKLangugage {
   TypeScript = "deno",
 }
 
+// with a round-robin load balancer emulation
+class TypegateManager {
+  private index = 0;
+
+  constructor(private typegates: Typegate[]) {}
+
+  get replicas() {
+    return this.typegates.length;
+  }
+
+  next() {
+    const typegate = this.typegates[this.index];
+    this.index = (this.index + 1) % this.typegates.length;
+    return typegate;
+  }
+
+  async terminate() {
+    await Promise.all(this.typegates.map((tg) => tg.deinit()));
+  }
+}
+
 interface ServeResult {
   port: number;
   cleanup: () => Promise<void>;
 }
 
-function serve(typegate: Typegate): Promise<ServeResult> {
+function serve(typegates: TypegateManager): Promise<ServeResult> {
   return new Promise((resolve) => {
     const server = Deno.serve({
       port: 0,
@@ -51,7 +72,7 @@ function serve(typegate: Typegate): Promise<ServeResult> {
         });
       },
     }, (req) => {
-      return typegate.handle(req, {
+      return typegates.next().handle(req, {
         remoteAddr: { hostname: "localhost" },
       } as Deno.ServeHandlerInfo);
     });
@@ -72,13 +93,13 @@ export class MetaTest {
 
   static async init(
     t: Deno.TestContext,
-    typegates: Typegate[],
+    typegates: TypegateManager,
     introspection: boolean,
     port = false,
   ): Promise<MetaTest> {
     const mt = new MetaTest(t, typegates, introspection);
     if (port) {
-      const { port: p, cleanup } = await serve(typegates[0]);
+      const { port: p, cleanup } = await serve(typegates);
       mt.port = p;
       mt.addCleanup(cleanup);
     }
@@ -88,20 +109,16 @@ export class MetaTest {
 
   private constructor(
     public t: Deno.TestContext,
-    public typegates: Typegate[],
+    public typegates: TypegateManager,
     private introspection: boolean,
   ) {}
 
   addCleanup(fn: MetaTestCleanupFn) {
     this.cleanups.push(fn);
   }
 
-  get typegate(): Typegate {
-    return this.typegates[this.currentTypegateIndex];
-  }
-
   getTypegraphEngine(name: string): QueryEngine | undefined {
-    return this.typegate.register.get(name);
+    return this.typegates.next().register.get(name);
   }
 
   async serialize(path: string, opts: ParseOptions = {}): Promise<string> {
@@ -140,7 +157,7 @@ export class MetaTest {
   }
 
   async undeploy(tgName: string) {
-    await this.typegate.register.remove(tgName);
+    await this.typegates.next().register.remove(tgName);
   }
 
   async engine(path: string, opts: ParseOptions = {}): Promise<QueryEngine> {
@@ -161,7 +178,7 @@ export class MetaTest {
       }
     }
 
-    const { engine, response } = await this.typegate.pushTypegraph(
+    const { engine, response } = await this.typegates.next().pushTypegraph(
       tgJson,
       secrets,
       this.introspection,
@@ -171,15 +188,12 @@ export class MetaTest {
       throw response.failure!;
     }
 
-    this.currentTypegateIndex += 1;
-    this.currentTypegateIndex %= this.typegates.length;
-
     return engine;
   }
 
   async engineFromDeployed(tgString: string): Promise<QueryEngine> {
     const tg = await TypeGraph.parseJson(tgString);
-    const { engine, response } = await this.typegate.pushTypegraph(
+    const { engine, response } = await this.typegates.next().pushTypegraph(
       tg,
       {},
       this.introspection,
@@ -211,20 +225,21 @@ export class MetaTest {
   }
 
   async unregister(engine: QueryEngine) {
+    const typegate = this.typegates.next();
     await Promise.all(
-      this.typegate.register
+      typegate.register
         .list()
         .filter((e) => e == engine)
         .map((e) => {
-          this.typegate.register.remove(e.name);
+          typegate.register.remove(e.name);
           return e.terminate();
         }),
     );
   }
 
   async terminate() {
     await Promise.all(this.cleanups.map((c) => c()));
-    await Promise.all(this.typegates.map((tg) => tg.deinit()));
+    await this.typegates.terminate();
   }
 
   async should(
@@ -294,7 +309,7 @@ interface TestConfig {
   // port on which the typegate instance will be exposed on expose the typegate instance
   port?: boolean;
   // number of typegate instances to create
-  multipleTypegates?: number;
+  replicas?: number;
   // create a temporary clean git repo for the tests
   gitRepo?: TempGitRepo;
   syncConfig?: SyncConfig;
@@ -321,26 +336,36 @@ export const test = ((o, fn): void => {
       if (opts.setup != null) {
         await opts.setup();
       }
-      const typegates: Typegate[] = [];
-      const typegate = await Typegate.init(opts.syncConfig ?? null);
-      typegates.push(typegate);
-      if (opts.multipleTypegates) {
-        for (let i = 0; i < opts.multipleTypegates - 1; i++) {
-          typegates.push(await Typegate.init(opts.syncConfig ?? null));
-        }
+      const replicas = opts.replicas ?? 1;
+      if (replicas < 1) {
+        throw new Error("replicas must be greater than 0");
+      }
+      if (replicas > 1 && opts.syncConfig == null) {
+        throw new Error(
+          "syncConfig must be provided when using multiple typegate instances",
+        );
       }
+
+      const typegates = await Promise.all(
+        Array.from({ length: replicas }).map((_) =>
+          Typegate.init(opts.syncConfig ?? null)
+        ),
+      );
+
       const {
         systemTypegraphs = false,
         gitRepo = null,
         introspection = false,
       } = opts;
       if (systemTypegraphs) {
-        await SystemTypegraph.loadAll(typegate);
+        await Promise.all(
+          typegates.map((typegate) => SystemTypegraph.loadAll(typegate)),
+        );
       }
 
       const mt = await MetaTest.init(
         t,
-        typegates,
+        new TypegateManager(typegates),
         introspection,
         opts.port != null,
       );