fix(satp-hermes): shutdown of SATP gateway should assure there are no pending transfers before shutting down

packages/cactus-plugin-satp-hermes/src/main/typescript/api1/dispatcher.ts

@@ -36,6 +36,7 @@ import type {
   ILocalLogRepository,
   IRemoteLogRepository,
 } from "../database/repository/interfaces/repository";
+import { GatewayShuttingDownError } from "./gateway-errors";
 
 export interface BLODispatcherOptions {
   logger: Logger;
@@ -65,6 +66,7 @@ export class BLODispatcher {
   private defaultRepository: boolean;
   private localRepository: ILocalLogRepository;
   private remoteRepository: IRemoteLogRepository | undefined;
+  private isShuttingDown = false;
 
   constructor(public readonly options: BLODispatcherOptions) {
     const fnTag = `${BLODispatcher.CLASS_NAME}#constructor()`;
@@ -198,9 +200,19 @@ export class BLODispatcher {
     return executeGetStatus(this.level, req, this.manager);
   }
 
+  /**
+   * @notice Transact request handler
+   * @param req TransactRequest
+   * @throws GatewayShuttingDownError when the flag isShuttingDown is true
+   * @returns TransactResponse
+  */
   public async Transact(req: TransactRequest): Promise<TransactResponse> {
     //TODO pre-verify verify input
+    const fnTag = `${BLODispatcher.CLASS_NAME}#transact()`;
     this.logger.info(`Transact request: ${req}`);
+    if (this.isShuttingDown) {
+      throw new GatewayShuttingDownError(fnTag);
+    }
     const res = await executeTransact(
       this.level,
       req,
@@ -215,6 +227,19 @@ export class BLODispatcher {
     const res = Array.from(await this.manager.getSessions().keys());
     return res;
   }
+
+  public async getManager(): Promise<SATPManager> {
+    this.logger.info(`Get SATP Manager request`);
+    return this.manager;
+  }
+
+  /**
+   * Changes the isShuttingDown flag to true, stopping all new requests
+  */
+  public setInitiateShutdown(): void {
+    this.logger.info(`Stopping requests`);
+    this.isShuttingDown = true;
+  }
   // get channel by caller; give needed client from orchestrator to handler to call
   // for all channels, find session id on request
   // TODO implement handlers GetAudit, Transact, Cancel, Routes

packages/cactus-plugin-satp-hermes/src/main/typescript/api1/gateway-errors.ts

@@ -0,0 +1,28 @@
+import { asError } from "@hyperledger/cactus-common";
+import { RuntimeError } from "run-time-error-cjs";
+import { Error as GatewayErrorType } from "../generated/proto/cacti/satp/v02/common/message_pb";
+export class GatewayError extends RuntimeError {
+  protected errorType = GatewayErrorType.UNSPECIFIED;
+  constructor(
+    public message: string,
+    public cause: string | Error | null,
+    // TODO internal error codes
+    public code: number = 500,
+    public traceID?: string,
+    public trace?: string,
+  ) {
+    super(message, asError(cause));
+    this.name = this.constructor.name;
+    Object.setPrototypeOf(this, new.target.prototype); // make sure prototype chain is set to error
+  }
+
+  public getGatewayErrorType(): GatewayErrorType {
+    return this.errorType;
+  }
+}
+
+export class GatewayShuttingDownError extends GatewayError {
+  constructor(tag: string, cause?: string | Error | null) {
+    super(`${tag}, shutdown initiated not receiving new requests`, cause ?? null, 500);
+  }
+}

packages/cactus-plugin-satp-hermes/src/main/typescript/core/satp-session.ts

@@ -1,6 +1,14 @@
 import { v4 as uuidv4 } from "uuid";
 import { stringify as safeStableStringify } from "safe-stable-stringify";
 
+import {
+  Checks,
+  JsObjectSigner,
+  LogLevelDesc,
+  Logger,
+  LoggerProvider,
+} from "@hyperledger/cactus-common";
+
 import {
   Type,
   MessageStagesHashesSchema,
@@ -55,6 +63,7 @@ import { create } from "@bufbuild/protobuf";
 
 // Define interface on protos
 export interface ISATPSessionOptions {
+  logLevel?: LogLevelDesc;
   contextID: string;
   sessionID?: string;
   server: boolean;
@@ -65,8 +74,16 @@ export class SATPSession {
   public static readonly CLASS_NAME = "SATPSession";
   private clientSessionData: SessionData | undefined;
   private serverSessionData: SessionData | undefined;
+  private readonly logger: Logger;
 
   constructor(ops: ISATPSessionOptions) {
+    const fnTag = `${SATPSession.CLASS_NAME}#constructor()`;
+    Checks.truthy(ops, `${fnTag} arg options`);
+
+    const level = ops.logLevel || "DEBUG";
+    const label = this.className;
+    this.logger = LoggerProvider.getOrCreate({ level, label });
+
     if (!ops.server && !ops.client) {
       throw new Error(`${SATPSession.CLASS_NAME}#constructor(), at least one of server or client must be true
     `);
@@ -135,6 +152,10 @@ export class SATPSession {
     return this.serverSessionData;
   }
 
+  public get className(): string {
+    return SATPSession.CLASS_NAME;
+  }
+
   public getClientSessionData(): SessionData {
     if (this.clientSessionData == undefined) {
       throw new Error(
@@ -219,11 +240,17 @@ export class SATPSession {
   }
 
   public getSessionId(): string {
-    console.log("serverSessionId: ", this.serverSessionData?.id);
-    console.log("clientSessionId: ", this.clientSessionData?.id);
+    this.logger.info("serverSessionId: ", this.serverSessionData?.state);
+    this.logger.info("clientSessionId: ", this.clientSessionData?.state);
     return this.serverSessionData?.id || this.clientSessionData?.id || "";
   }
 
+  public getSessionState(): State {
+    this.logger.info("serverSessionId: ", this.serverSessionData?.state);
+    this.logger.info("clientSessionId: ", this.clientSessionData?.state);
+    return this.serverSessionData?.state || this.clientSessionData?.state || State.UNSPECIFIED;
+  }
+
   public verify(
     tag: string,
     type: SessionType,

packages/cactus-plugin-satp-hermes/src/main/typescript/plugin-satp-hermes-gateway.ts

@@ -65,6 +65,7 @@ import cors from "cors";
 import * as OAS from "../json/openapi-blo-bundled.json";
 import type { NetworkId } from "./services/network-identification/chainid-list";
 import { knexLocalInstance } from "./database/knexfile";
+import schedule, { Job } from "node-schedule";
 
 export class SATPGateway implements IPluginWebService, ICactusPlugin {
   @IsDefined()
@@ -104,6 +105,8 @@ export class SATPGateway implements IPluginWebService, ICactusPlugin {
   public remoteRepository?: IRemoteLogRepository;
   private readonly shutdownHooks: ShutdownHook[];
   private crashManager?: CrashManager;
+  private sessionVerificationJob: Job | null = null;
+
 
   constructor(public readonly options: SATPGatewayConfig) {
     const fnTag = `${this.className}#constructor()`;
@@ -583,7 +586,7 @@ export class SATPGateway implements IPluginWebService, ICactusPlugin {
   public async shutdown(): Promise<void> {
     const fnTag = `${this.className}#getGatewaySeeds()`;
     this.logger.debug(`Entering ${fnTag}`);
-
+    
     this.logger.info("Shutting down Node server - BOL");
     await this.shutdownBLOServer();
     await this.shutdownGOLServer();
@@ -618,6 +621,7 @@ export class SATPGateway implements IPluginWebService, ICactusPlugin {
     this.logger.debug(`Entering ${fnTag}`);
     if (this.BLOServer) {
       try {
+        await this.verifySessionsState();
         await this.BLOServer.closeAllConnections();
         await this.BLOServer.close();
         this.BLOServer = undefined;
@@ -649,4 +653,51 @@ export class SATPGateway implements IPluginWebService, ICactusPlugin {
       this.logger.warn("Server is not running.");
     }
   }
-}
+
+  /**
+   * Verify the state of the sessions before shutting down the server.
+   * This method is called before the server is shut down and awaits ensure that 
+   * all sessions are concluded before the server is terminated.
+   * After all sessions are concluded, the job is cancelled.
+  */
+  private async verifySessionsState(): Promise<void> {
+    const fnTag = `${this.className}#verifySessionsState()`;
+    this.logger.trace(`Entering ${fnTag}`);
+    if (!this.BLODispatcher) {
+      throw new Error(`Cannot ${fnTag} because BLODispatcher is erroneous`);
+    }
+    this.BLODispatcher.setInitiateShutdown();
+    const manager = await this.BLODispatcher.getManager();
+
+    await this.startSessionVerificationJob(manager);
+    this.logger.info("Session verification process started.");
+  }
+
+
+  /**
+   * Start a scheduled job to verify session states.
+   * The job runs every 20 seconds until all sessions are concluded.
+  */
+  private async startSessionVerificationJob(manager: any): Promise<void> {
+    const fnTag = `${this.className}#startSessionVerificationJob()`;
+    this.logger.trace(`Entering ${fnTag}`);
+
+    this.sessionVerificationJob = schedule.scheduleJob("*/20 * * * * *", async () => {
+      try {
+        const status = await manager.getSATPSessionState();
+        if (!status) {
+          this.logger.info("Sessions are still pending");
+        } else {
+          this.logger.info("All sessions are concluded");
+          if (this.sessionVerificationJob) {
+            this.sessionVerificationJob.cancel(); // Stop the job once sessions are concluded
+            this.logger.info("Session verification job stopped.");
+          }
+        }
+      } catch (error) {
+        this.logger.error(`Error in session verification job: ${error}`);
+      }
+    });
+  }
+
+}

packages/cactus-plugin-satp-hermes/src/main/typescript/services/gateway/satp-manager.ts

@@ -37,6 +37,7 @@ import { Stage2SATPHandler } from "../../core/stage-handlers/stage2-handler";
 import { Stage3SATPHandler } from "../../core/stage-handlers/stage3-handler";
 import { SATPCrossChainManager } from "../../cross-chain-mechanisms/satp-cc-manager";
 import { GatewayOrchestrator } from "./gateway-orchestrator";
+import { State } from "../../generated/proto/cacti/satp/v02/common/session_pb";
 import type { SessionData } from "../../generated/proto/cacti/satp/v02/common/session_pb";
 import type { SatpStage0Service } from "../../generated/proto/cacti/satp/v02/service/stage_0_pb";
 import type { SatpStage1Service } from "../../generated/proto/cacti/satp/v02/service/stage_1_pb";
@@ -310,6 +311,21 @@ export class SATPManager {
     return this.satpHandlers.get(type);
   }
 
+  /*
+    * Function checks if all the sessions are in the completed state
+    * @returns boolean
+  */
+  public getSATPSessionState(): boolean {
+    const fnTag = `${SATPManager.CLASS_NAME}#getSATPSessionStatus()`;
+    this.logger.info(`${fnTag}, Getting SATP Session Status...`);
+    for (let value of this.sessions.values()) {
+      if (value.getSessionState() !== State.COMPLETED) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   public getOrCreateSession(
     sessionId?: string,
     contextID?: string,