feat(spanner): executor framework

handwritten/spanner/google-cloud-spanner-executor/src/cloud-client-executor.ts

@@ -0,0 +1,254 @@
+/*!
+ * Copyright 2026 Google LLC. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ServerDuplexStream, status } from '@grpc/grpc-js';
+import { Spanner } from '../../src';
+import { trace, context, Tracer } from '@opentelemetry/api';
+import * as protos from '../../protos/protos';
+import { CloudUtil } from './cloud-util';
+import { OutcomeSender, IExecutionFlowContext } from './cloud-executor';
+import spanner = protos.google.spanner;
+import SpannerAsyncActionRequest = spanner.executor.v1.SpannerAsyncActionRequest;
+import SpannerAsyncActionResponse = spanner.executor.v1.SpannerAsyncActionResponse;
+import ISpannerAction = spanner.executor.v1.ISpannerAction;
+import IAdminAction = spanner.executor.v1.IAdminAction;
+import ICreateCloudInstanceAction = spanner.executor.v1.ICreateCloudInstanceAction;
+
+/**
+ * Context for a single stream connection.
+ */
+export class ExecutionFlowContext implements IExecutionFlowContext {
+  private call: ServerDuplexStream<
+    SpannerAsyncActionRequest,
+    SpannerAsyncActionResponse
+  >;
+  private dbPath: string = '';
+
+  constructor(
+    call: ServerDuplexStream<
+      SpannerAsyncActionRequest,
+      SpannerAsyncActionResponse
+    >
+  ) {
+    this.call = call;
+  }
+
+  /**
+   * Sends a response back to the client.
+   */
+  public onNext(response: SpannerAsyncActionResponse): void {
+    this.call.write(response);
+  }
+
+  /**
+   * Sends an error back to the client.
+   */
+  public onError(error: Error): void {
+    this.call.emit('error', error);
+  }
+
+  /**
+   * Clean up resources associated with the context.
+  */
+  public cleanup(): void {
+    console.log('Cleaning up ExecutionFlowContext');
+  }
+
+  /**
+   * Gets or sets the database path for the context.
+   */
+  public getDatabasePath(path: string | null | undefined): string {
+    if (!path) {
+      return this.dbPath;
+    }
+    this.dbPath = path;
+    return path;
+  }
+}
+/**
+ * CloudClientExecutor handles the execution of Spanner actions requested via the executor proxy.
+*/
+export class CloudClientExecutor {
+  private spanner: Spanner;
+  private tracer: Tracer;
+  private enableGrpcFaultInjector: boolean;
+
+  constructor(enableGrpcFaultInjector: boolean) {
+    this.enableGrpcFaultInjector = enableGrpcFaultInjector;
+    const spannerOptions = CloudUtil.getSpannerOptions();
+    this.spanner = new Spanner(spannerOptions);
+    this.tracer = trace.getTracer(CloudClientExecutor.name);
+  }
+
+  /**
+   * Creates a new ExecutionFlowContext for a stream.
+  */
+  public createExecutionFlowContext(
+    call: ServerDuplexStream<
+      SpannerAsyncActionRequest,
+      SpannerAsyncActionResponse
+    >
+  ): ExecutionFlowContext {
+    return new ExecutionFlowContext(call);
+  }
+
+  /**
+   * Starts handling a SpannerAsyncActionRequest.
+   */
+  public async startHandlingRequest(
+    req: SpannerAsyncActionRequest,
+    executionContext: ExecutionFlowContext
+  ): Promise<{ code: number; details: string }> {
+    const outcomeSender = new OutcomeSender(req.actionId!, executionContext);
+
+    if (!req.action) {
+      return outcomeSender.finishWithError({
+        code: status.INVALID_ARGUMENT,
+        message: 'Invalid request: No action present',
+      });
+    }
+
+    const action = req.action;
+    const dbPath = executionContext.getDatabasePath(action.databasePath);
+
+    let useMultiplexedSession = false;
+    if (action.spannerOptions?.sessionPoolOptions) {
+      useMultiplexedSession = !!action.spannerOptions.sessionPoolOptions.useMultiplexed;
+    }
+    /**
+      * Executes the appropriate action based on the request.
+    */
+    this.executeAction(
+      outcomeSender,
+      action,
+      dbPath,
+      useMultiplexedSession,
+      executionContext
+    ).catch(err => {
+      console.error('Unhandled exception in action execution:', err);
+      outcomeSender.finishWithError(err);
+    });
+
+    return { code: status.OK, details: '' };
+  }
+  /**
+   * Executes the CreateCloudInstance action.
+  */
+  private async executeAction(
+    outcomeSender: OutcomeSender,
+    action: ISpannerAction,
+    dbPath: string,
+    useMultiplexedSession: boolean,
+    executionContext: ExecutionFlowContext
+  ): Promise<{ code: number; details: string }> {
+    const actionType = Object.keys(action).find(k => (action as any)[k] !== undefined) || 'unknown';
+    const span = this.tracer.startSpan(`performaction_${actionType}`);
+
+    return context.with(trace.setSpan(context.active(), span), async () => {
+      try {
+        if (action.admin) {
+          return await this.executeAdminAction(action.admin, outcomeSender);
+        }
+
+        return outcomeSender.finishWithError({
+          code: status.UNIMPLEMENTED,
+          message: `Action ${actionType} not implemented yet`,
+        });
+      } catch (e: any) {
+        span.recordException(e);
+        console.warn('Unexpected error:', e);
+        return outcomeSender.finishWithError({
+          code: status.INVALID_ARGUMENT,
+          message: `Unexpected error: ${e.message}`,
+        });
+      } finally {
+        span.end();
+      }
+    });
+  }
+
+  private async executeAdminAction(
+    action: IAdminAction,
+    sender: OutcomeSender
+  ): Promise<{ code: number; details: string }> {
+    try {
+      if (action.createCloudInstance) {
+        return await this.executeCreateCloudInstance(action.createCloudInstance, sender);
+      }
+      return sender.finishWithError({
+        code: status.UNIMPLEMENTED,
+        message: 'Admin action not implemented',
+      });
+    } catch (e: any) {
+      return sender.finishWithError(e);
+    }
+  }
+
+  private async executeCreateCloudInstance(
+    action: ICreateCloudInstanceAction,
+    sender: OutcomeSender
+  ): Promise<{ code: number; details: string }> {
+    try {
+      console.log(`Creating instance: \n${JSON.stringify(action, null, 2)}`);
+
+      const instanceId = action.instanceId!;
+      const projectId = action.projectId!;
+      const configId = action.instanceConfigId!;
+
+      const instanceAdminClient = this.spanner.getInstanceAdminClient();
+
+      const [operation] = await instanceAdminClient.createInstance({
+        parent: instanceAdminClient.projectPath(projectId),
+        instanceId: instanceId,
+        instance: {
+          config: instanceAdminClient.instanceConfigPath(projectId, configId),
+          displayName: instanceId,
+          nodeCount: action.nodeCount || 1,
+          processingUnits: action.processingUnits,
+          labels: action.labels || {},
+        },
+      });
+
+      console.log('Waiting for instance creation operation to complete...');
+
+      await operation.promise();
+
+      console.log(`Instance ${instanceId} created successfully.`);
+
+      return sender.finishWithOK();
+    } catch (err: any) {
+      if (err.code === status.ALREADY_EXISTS) {
+        console.log('Instance already exists, returning OK.');
+        return sender.finishWithOK();
+      }
+      console.error('Failed to create instance:', err);
+      return sender.finishWithError(err);
+    }
+  }
+  /**
+   * Simulates an end-to-end trace verification task.
+  */
+  public async getEndToEndTraceVerificationTask(traceId: string): Promise<boolean> {
+    return new Promise(resolve => {
+      setTimeout(async () => {
+        console.log(`Verifying trace ${traceId}...`);
+        resolve(true);
+      }, 10000);
+    });
+  }
+}
+
+