test: add corestore volume scenario

package.json

@@ -5,9 +5,10 @@
   "type": "module",
   "description": "Whole-system integration scenarios for serve.zone components.",
   "scripts": {
-    "bootstrap:components": "pnpm --dir ../interfaces install && pnpm --dir ../api install && pnpm --dir ../cloudly install && pnpm --dir ../coreflow install && pnpm --dir ../coretraffic install && pnpm --dir ../onebox exec deno install --config deno.json && pnpm install",
+    "bootstrap:components": "pnpm --dir ../interfaces install && pnpm --dir ../api install && pnpm --dir ../cloudly install && pnpm --dir ../coreflow install && pnpm --dir ../coretraffic install && pnpm --dir ../corestore install && pnpm --dir ../onebox exec deno install --config deno.json && pnpm install",
-    "test": "pnpm scenario:registry-deploy-on-push && pnpm scenario:onebox-basic-lifecycle && pnpm scenario:onebox-backup-restore",
+    "test": "pnpm scenario:corestore-volume-driver && pnpm scenario:registry-deploy-on-push && pnpm scenario:onebox-basic-lifecycle && pnpm scenario:onebox-backup-restore",
     "test:full": "pnpm test && pnpm scenario:onebox-cloudly-appstore-worker",
+    "scenario:corestore-volume-driver": "tsx scenarios/corestore-volume-driver/scenario.ts",
     "scenario:registry-deploy-on-push": "tsx --tsconfig ../cloudly/tsconfig.json scenarios/registry-deploy-on-push/scenario.ts",
     "scenario:onebox-basic-lifecycle": "deno run --allow-all --config ../onebox/deno.json scenarios/onebox-basic-lifecycle/scenario.ts",
     "scenario:onebox-backup-restore": "deno run --allow-all --config ../onebox/deno.json scenarios/onebox-backup-restore/scenario.ts",

readme.md

@@ -7,6 +7,7 @@ Fast package tests stay in each component repo. This repo is for stateful cross-
 ## Scenarios
 
 - `registry-deploy-on-push`: starts Cloudly with isolated Mongo/S3 helpers, connects Coreflow as a cluster, pushes a Docker image to Cloudly's built-in registry, verifies Cloudly metadata updates, verifies Coreflow creates the workload service, verifies Coretraffic HTTPS routing, then pushes the same tag again and verifies service recreation and routing through the new digest.
+- `corestore-volume-driver`: starts Corestore with isolated ports and a temporary Docker plugin socket, verifies the VolumeDriver protocol, snapshots/restores a managed volume through `containerarchive`, and checks Coreflow's generated corestore volume mount specs.
 - `onebox-basic-lifecycle`: starts Onebox in dev mode, verifies core services, deploys a workload, checks HTTP plus HTTPS routing through ingress, removes the workload, and verifies cleanup.
 
 ## Host Run

scenarios/corestore-volume-driver/scenario.ts

@@ -0,0 +1,335 @@
+import * as http from 'node:http';
+import * as net from 'node:net';
+import { access, mkdir, readFile, rm, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { dirname, join } from 'node:path';
+
+import { CoreStore } from '../../../corestore/ts/index.js';
+import { Coreflow } from '../../../coreflow/ts/coreflow.classes.coreflow.js';
+import type { IService } from '../../../interfaces/ts/data/service.js';
+
+const scenarioName = 'corestore-volume-driver';
+const smokeId = `corestore-volume-${Date.now().toString(36)}`;
+const buildDir = join(tmpdir(), smokeId);
+const apiToken = `${smokeId}-token`;
+
+function assert(conditionArg: unknown, messageArg: string): asserts conditionArg {
+  if (!conditionArg) {
+    throw new Error(messageArg);
+  }
+}
+
+const getFreePort = async () => {
+  return await new Promise<number>((resolveArg, rejectArg) => {
+    const server = net.createServer();
+    server.on('error', rejectArg);
+    server.listen(0, '127.0.0.1', () => {
+      const address = server.address();
+      if (typeof address !== 'object' || !address) {
+        server.close(() => rejectArg(new Error('Could not allocate a port')));
+        return;
+      }
+      server.close(() => resolveArg(address.port));
+    });
+  });
+};
+
+const requestJson = async <T>(optionsArg: http.RequestOptions, bodyArg?: unknown) => {
+  return await new Promise<T>((resolveArg, rejectArg) => {
+    const payload = bodyArg === undefined ? '' : JSON.stringify(bodyArg);
+    const headers = {
+      ...(bodyArg === undefined
+        ? {}
+        : {
+            'content-type': 'application/json',
+            'content-length': Buffer.byteLength(payload),
+          }),
+      ...(optionsArg.headers || {}),
+    };
+    const request = http.request(
+      {
+        timeout: 30000,
+        ...optionsArg,
+        headers,
+      },
+      (response) => {
+        const chunks: Buffer[] = [];
+        response.on('data', (chunkArg) => {
+          chunks.push(Buffer.isBuffer(chunkArg) ? chunkArg : Buffer.from(chunkArg));
+        });
+        response.on('end', () => {
+          const responseText = Buffer.concat(chunks).toString('utf8');
+          if ((response.statusCode || 500) >= 400) {
+            rejectArg(new Error(`HTTP ${response.statusCode}: ${responseText}`));
+            return;
+          }
+          resolveArg(responseText ? JSON.parse(responseText) as T : ({} as T));
+        });
+      },
+    );
+    request.on('timeout', () => request.destroy(new Error(`Request timed out: ${optionsArg.path}`)));
+    request.on('error', rejectArg);
+    request.end(payload);
+  });
+};
+
+const controlGet = async <T>(controlPortArg: number, pathArg: string) => {
+  return await requestJson<T>({
+    method: 'GET',
+    hostname: '127.0.0.1',
+    port: controlPortArg,
+    path: pathArg,
+    headers: {
+      authorization: `Bearer ${apiToken}`,
+    },
+  });
+};
+
+const controlPost = async <T>(controlPortArg: number, pathArg: string, bodyArg: unknown) => {
+  return await requestJson<T>(
+    {
+      method: 'POST',
+      hostname: '127.0.0.1',
+      port: controlPortArg,
+      path: pathArg,
+      headers: {
+        authorization: `Bearer ${apiToken}`,
+      },
+    },
+    bodyArg,
+  );
+};
+
+const pluginPost = async <T>(socketPathArg: string, pathArg: string, bodyArg: unknown) => {
+  return await requestJson<T>(
+    {
+      method: 'POST',
+      socketPath: socketPathArg,
+      path: pathArg,
+    },
+    bodyArg,
+  );
+};
+
+const pathExists = async (pathArg: string) => {
+  try {
+    await access(pathArg);
+    return true;
+  } catch {
+    return false;
+  }
+};
+
+const assertCoreflowVolumeMounts = () => {
+  const coreflow = new Coreflow();
+  const service: IService = {
+    id: 'svc-corestore-volume-test',
+    data: {
+      name: 'app.volume-test',
+      description: 'Corestore volume mount spec test service',
+      imageId: 'image-corestore-volume-test',
+      imageVersion: 'latest',
+      environment: {},
+      secretBundleId: 'secret-corestore-volume-test',
+      serviceCategory: 'workload',
+      deploymentStrategy: 'custom',
+      scaleFactor: 1,
+      balancingStrategy: 'round-robin',
+      ports: {
+        web: 80,
+      },
+      volumes: [
+        {
+          mountPath: '/data',
+          backup: true,
+        },
+        {
+          source: 'shared-assets',
+          mountPath: '/assets',
+          readOnly: true,
+          options: {
+            tier: 'gold',
+          },
+        },
+      ],
+      resources: {
+        memorySizeLimitMB: 128,
+        volumeMounts: [
+          {
+            hostFsPath: '/tmp/legacy-corestore-volume-test',
+            containerFsPath: '/legacy',
+          },
+        ],
+      },
+      domains: [],
+      deploymentIds: [],
+    },
+  };
+
+  const clusterManager = coreflow.clusterManager as unknown as {
+    getServiceDockerMounts: (serviceArg: IService) => Array<Record<string, any>>;
+    getServiceVolumeHash: (serviceArg: IService) => string;
+  };
+  const mounts = clusterManager.getServiceDockerMounts(service);
+  const volumeHash = clusterManager.getServiceVolumeHash(service);
+
+  assert(volumeHash.length === 64, `Expected non-empty volume hash, got ${volumeHash}`);
+  assert(mounts.some((mountArg) => mountArg.Type === 'bind' && mountArg.Target === '/legacy'), 'Legacy bind mount was not preserved');
+
+  const dataMount = mounts.find((mountArg) => mountArg.Type === 'volume' && mountArg.Target === '/data');
+  assert(dataMount, `Missing /data volume mount: ${JSON.stringify(mounts)}`);
+  assert(dataMount.VolumeOptions.DriverConfig.Name === 'corestore', 'Default volume driver should be corestore');
+  assert(dataMount.VolumeOptions.DriverConfig.Options.serviceId === service.id, 'Missing serviceId driver option');
+  assert(dataMount.VolumeOptions.DriverConfig.Options.serviceName === service.data.name, 'Missing serviceName driver option');
+  assert(dataMount.VolumeOptions.DriverConfig.Options.mountPath === '/data', 'Missing mountPath driver option');
+  assert(dataMount.VolumeOptions.DriverConfig.Options.backup === 'true', 'Missing backup driver option');
+  assert(typeof dataMount.Source === 'string' && dataMount.Source.startsWith('sz-app.volume-test-data-'), `Unexpected generated volume name: ${dataMount.Source}`);
+
+  const sharedMount = mounts.find((mountArg) => mountArg.Type === 'volume' && mountArg.Target === '/assets');
+  assert(sharedMount, `Missing /assets volume mount: ${JSON.stringify(mounts)}`);
+  assert(sharedMount.Source === 'shared-assets', 'Explicit volume source was not preserved');
+  assert(sharedMount.ReadOnly === true, 'Read-only flag was not preserved');
+  assert(sharedMount.VolumeOptions.DriverConfig.Options.tier === 'gold', 'Custom driver option was not preserved');
+
+  const serviceWithoutVolumes: IService = {
+    ...service,
+    data: {
+      ...service.data,
+      volumes: [],
+      resources: undefined,
+    },
+  };
+  assert(clusterManager.getServiceVolumeHash(serviceWithoutVolumes) === '', 'Services without volumes should not get a volume hash label');
+};
+
+const assertCorestoreVolumeDriver = async () => {
+  const socketPath = join(buildDir, 'plugins', 'corestore.sock');
+  const dataDir = join(buildDir, 'data');
+  await mkdir(dirname(socketPath), { recursive: true });
+
+  const controlPort = await getFreePort();
+  const s3Port = await getFreePort();
+  const dbPort = await getFreePort();
+  const corestore = new CoreStore({
+    dataDir,
+    bindAddress: '127.0.0.1',
+    publicHost: '127.0.0.1',
+    controlPort,
+    s3Port,
+    dbPort,
+    apiToken,
+    volumePluginSocketPath: socketPath,
+  });
+
+  try {
+    await corestore.start();
+
+    const health = await controlGet<any>(controlPort, '/health');
+    assert(health.ok === true, `Corestore health was not ok: ${JSON.stringify(health)}`);
+    assert(health.volumes?.running === true, `Volume plugin health was not reported: ${JSON.stringify(health)}`);
+
+    const activate = await pluginPost<any>(socketPath, '/Plugin.Activate', {});
+    assert(activate.Implements?.includes('VolumeDriver'), `Plugin activation failed: ${JSON.stringify(activate)}`);
+
+    const capabilities = await pluginPost<any>(socketPath, '/VolumeDriver.Capabilities', {});
+    assert(capabilities.Capabilities?.Scope === 'local', `Unexpected plugin capabilities: ${JSON.stringify(capabilities)}`);
+
+    const create = await pluginPost<any>(socketPath, '/VolumeDriver.Create', {
+      Name: 'smoke-volume',
+      Opts: {
+        serviceId: 'svc-smoke',
+        serviceName: 'smoke-service',
+        mountPath: '/data',
+        backup: 'true',
+      },
+    });
+    assert(create.Err === '', `Volume create failed: ${JSON.stringify(create)}`);
+
+    const mount = await pluginPost<any>(socketPath, '/VolumeDriver.Mount', {
+      Name: 'smoke-volume',
+      ID: 'container-smoke',
+    });
+    assert(mount.Err === '' && typeof mount.Mountpoint === 'string', `Volume mount failed: ${JSON.stringify(mount)}`);
+
+    const removeWhileMounted = await pluginPost<any>(socketPath, '/VolumeDriver.Remove', {
+      Name: 'smoke-volume',
+    });
+    assert(typeof removeWhileMounted.Err === 'string' && removeWhileMounted.Err.includes('still mounted'), 'Mounted volume removal should be rejected');
+
+    const pathResponse = await pluginPost<any>(socketPath, '/VolumeDriver.Path', {
+      Name: 'smoke-volume',
+    });
+    assert(pathResponse.Mountpoint === mount.Mountpoint, `Volume path mismatch: ${JSON.stringify(pathResponse)}`);
+
+    const getResponse = await pluginPost<any>(socketPath, '/VolumeDriver.Get', {
+      Name: 'smoke-volume',
+    });
+    assert(getResponse.Volume?.Status?.serviceId === 'svc-smoke', `Volume metadata missing: ${JSON.stringify(getResponse)}`);
+
+    const pluginList = await pluginPost<any>(socketPath, '/VolumeDriver.List', {});
+    assert(pluginList.Volumes?.some((volumeArg: any) => volumeArg.Name === 'smoke-volume'), `Plugin list missing volume: ${JSON.stringify(pluginList)}`);
+
+    const controlList = await controlGet<any>(controlPort, '/volumes');
+    assert(controlList.volumes?.some((volumeArg: any) => volumeArg.Name === 'smoke-volume'), `Control list missing volume: ${JSON.stringify(controlList)}`);
+
+    await mkdir(join(mount.Mountpoint, 'nested'), { recursive: true });
+    await writeFile(join(mount.Mountpoint, 'nested', 'hello.txt'), 'corestore volume smoke\n');
+
+    const snapshotResponse = await controlPost<any>(controlPort, '/volumes/snapshot', {
+      name: 'smoke-volume',
+      snapshotName: 'before-mutation',
+      tags: {
+        scenario: scenarioName,
+        smokeId,
+      },
+    });
+    assert(snapshotResponse.snapshot?.id, `Snapshot missing id: ${JSON.stringify(snapshotResponse)}`);
+
+    await writeFile(join(mount.Mountpoint, 'stale.txt'), 'should be removed on restore\n');
+    await writeFile(join(mount.Mountpoint, 'nested', 'hello.txt'), 'mutated\n');
+
+    const snapshotList = await controlGet<any>(controlPort, '/volumes/snapshots?name=smoke-volume');
+    assert(snapshotList.snapshots?.some((snapshotArg: any) => snapshotArg.snapshotId === snapshotResponse.snapshot.id), `Snapshot list missing snapshot: ${JSON.stringify(snapshotList)}`);
+
+    const restoreResponse = await controlPost<any>(controlPort, '/volumes/restore', {
+      name: 'smoke-volume',
+      snapshotId: snapshotResponse.snapshot.id,
+    });
+    assert(restoreResponse.ok === true, `Restore failed: ${JSON.stringify(restoreResponse)}`);
+
+    const restoredContent = await readFile(join(mount.Mountpoint, 'nested', 'hello.txt'), 'utf8');
+    assert(restoredContent === 'corestore volume smoke\n', `Restore content mismatch: ${restoredContent}`);
+    assert(!(await pathExists(join(mount.Mountpoint, 'stale.txt'))), 'Restore did not clear stale files');
+
+    const unmount = await pluginPost<any>(socketPath, '/VolumeDriver.Unmount', {
+      Name: 'smoke-volume',
+      ID: 'container-smoke',
+    });
+    assert(unmount.Err === '', `Volume unmount failed: ${JSON.stringify(unmount)}`);
+
+    const remove = await pluginPost<any>(socketPath, '/VolumeDriver.Remove', {
+      Name: 'smoke-volume',
+    });
+    assert(remove.Err === '', `Volume remove failed: ${JSON.stringify(remove)}`);
+
+    const finalList = await pluginPost<any>(socketPath, '/VolumeDriver.List', {});
+    assert(!finalList.Volumes?.some((volumeArg: any) => volumeArg.Name === 'smoke-volume'), `Volume was not removed: ${JSON.stringify(finalList)}`);
+  } finally {
+    await corestore.stop().catch((errorArg) => {
+      console.log(`[${scenarioName}] Failed to stop Corestore: ${(errorArg as Error).message}`);
+    });
+  }
+};
+
+const main = async () => {
+  try {
+    await mkdir(buildDir, { recursive: true });
+    assertCoreflowVolumeMounts();
+    await assertCorestoreVolumeDriver();
+    console.log(`[${scenarioName}] PASS`);
+  } finally {
+    await rm(buildDir, { recursive: true, force: true });
+  }
+};
+
+await main();