fix: surface mid-flight error identity, liveness, observability (#1-#6,#10-#13)

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@moonlight-protocol/provider-platform",
-  "version": "0.9.7",
+  "version": "0.9.8",
   "license": "MIT",
   "exports": "./src/main.ts",
   "tasks": {
@@ -27,7 +27,7 @@
     "@colibri/core": "jsr:@colibri/core@^0.22.0",
     "@drizzle-team/brocli": "npm:@drizzle-team/brocli@^0.11.0",
     "@fifo/convee": "jsr:@fifo/convee@^0.10.0",
-    "@moonlight/moonlight-sdk": "jsr:@moonlight/moonlight-sdk@^0.11.0",
+    "@moonlight/moonlight-sdk": "jsr:@moonlight/moonlight-sdk@^0.11.1",
     "@std/assert": "jsr:@std/assert@^1.0.0",
     "@std/dotenv": "jsr:@std/dotenv@^0.225.6",
     "@zaubrik/djwt": "jsr:@zaubrik/djwt@^3.0.2",

src/core/service/auth/challenge/verify/verify-challenge.ts

@@ -4,6 +4,7 @@ import { getProviderAccount } from "@/core/service/auth/service/service-account.
 import { NETWORK_CONFIG, SERVICE_DOMAIN } from "@/config/env.ts";
 import type { PostChallengeInput } from "@/core/service/auth/challenge/types.ts";
 import * as E from "@/core/service/auth/challenge/verify/error.ts";
+import { PlatformError } from "@/error/index.ts";
 import { assertOrThrow } from "@/utils/error/assert-or-throw.ts";
 import { isDefined } from "@/utils/type-guards/is-defined.ts";
 import { StrKey } from "@colibri/core";
@@ -128,6 +129,10 @@ export const P_VerifyChallenge = (deps: { log: Logger }) =>
               : String(error),
           });
           log.error(error, "challenge verification failed");
+          // Preserve a precise auth failure (e.g. MISSING_CLIENT_SIGNATURE
+          // 4xx) instead of masking every cause as a generic 500 (#4). Only
+          // genuinely unexpected errors become CHALLENGE_VERIFICATION_FAILED.
+          if (error instanceof PlatformError) throw error;
           throw new E.CHALLENGE_VERIFICATION_FAILED(error);
         }
       });

src/core/service/bundle/add-bundle.process.ts

@@ -8,6 +8,7 @@ import type { PostEndpointInput } from "@/http/pipelines/types.ts";
 import type { OperationTypes } from "@moonlight/moonlight-sdk";
 import { MoonlightOperation } from "@moonlight/moonlight-sdk";
 import { resolveChannelContext } from "@/core/service/executor/channel-resolver.ts";
+import { TimeoutError, withTimeout } from "@/utils/async/with-timeout.ts";
 import {
   calculateBundleTtl,
   calculateBundleWeight,
@@ -54,6 +55,10 @@ const operationsBundleRepository = new OperationsBundleRepository(
   drizzleClient,
 );
 
+// Bounded so an unreachable Soroban RPC fast-fails admission (#2) instead of
+// hanging the request for minutes.
+const ADMISSION_RPC_TIMEOUT_MS = 10_000;
+
 const MEMPOOL_WEIGHT_CONFIG: WeightConfig = {
   expensiveOpWeight: MEMPOOL_EXPENSIVE_OP_WEIGHT,
   cheapOpWeight: MEMPOOL_CHEAP_OP_WEIGHT,
@@ -137,16 +142,13 @@ function parseOperations(
       "operations.count": operationsMLXDR.length,
     });
     log.event("parsing MLXDR operations");
+    // `operationsMLXDR` is a 1:1 source for `operations` and the request
+    // schema already enforces `.min(1)` (returning a structured 400 before
+    // this runs), so an empty-operations case is unreachable here (#13).
     const operations = await Promise.all(
       operationsMLXDR.map((xdr) => MoonlightOperation.fromMLXDR(xdr)),
     );
 
-    if (operations.length === 0) {
-      span.addEvent("no_operations");
-      log.event("no operations parsed");
-      throw new E.NO_OPERATIONS_PROVIDED();
-    }
-
     span.addEvent("operations_parsed", {
       "operations.count": operations.length,
     });
@@ -210,6 +212,7 @@ function persistSpendOperations(
   bundleId: string,
   accountId: string,
   channelClient: import("@moonlight/moonlight-sdk").PrivacyChannel,
+  channelContractId: string,
   deps: { log: Logger },
 ): Promise<void> {
   if (operations.length === 0) {
@@ -223,11 +226,16 @@ function persistSpendOperations(
     });
 
     const utxoPublicKeys = operations.map((op) => op.getUtxo());
-    const balances = await fetchUtxoBalances(
-      utxoPublicKeys,
-      channelClient,
-      deps,
-    );
+    const balances = await withTimeout(
+      fetchUtxoBalances(utxoPublicKeys, channelClient, deps),
+      ADMISSION_RPC_TIMEOUT_MS,
+      "fetchUtxoBalances",
+    ).catch((error) => {
+      if (error instanceof TimeoutError) {
+        throw new E.CHANNEL_RPC_UNAVAILABLE(channelContractId);
+      }
+      throw error;
+    });
 
     for (let i = 0; i < operations.length; i++) {
       const operation = operations[i];
@@ -350,11 +358,18 @@ export const P_AddOperationsBundle = (deps: { log: Logger }) =>
         log.debug("ppPublicKey", ppPublicKey);
 
         log.event("resolving channel context for PP");
-        const channelCtx = await resolveChannelContext(
-          channelContractId,
-          ppPublicKey,
-          deps,
-        );
+        const channelCtx = await withTimeout(
+          resolveChannelContext(channelContractId, ppPublicKey, deps),
+          ADMISSION_RPC_TIMEOUT_MS,
+          "resolveChannelContext",
+        ).catch((error) => {
+          // Fast-fail an unreachable chain during admission instead of
+          // hanging the request indefinitely (#2).
+          if (error instanceof TimeoutError) {
+            throw new E.CHANNEL_RPC_UNAVAILABLE(channelContractId);
+          }
+          throw error;
+        });
         const channelClient = channelCtx.channelClient;
 
         span.addEvent("validating_session");
@@ -432,11 +447,18 @@ export const P_AddOperationsBundle = (deps: { log: Logger }) =>
 
         span.addEvent("calculating_fee");
         log.event("calculating fee");
-        const amounts = await calculateOperationAmounts(
-          classified,
-          channelClient,
-          deps,
-        );
+        // First on-chain read of admission — fast-fail if the chain is
+        // unreachable rather than hanging the request (#2).
+        const amounts = await withTimeout(
+          calculateOperationAmounts(classified, channelClient, deps),
+          ADMISSION_RPC_TIMEOUT_MS,
+          "calculateOperationAmounts",
+        ).catch((error) => {
+          if (error instanceof TimeoutError) {
+            throw new E.CHANNEL_RPC_UNAVAILABLE(channelContractId);
+          }
+          throw error;
+        });
         const feeCalculation = calculateFee(amounts);
 
         span.addEvent("fee_calculated", {
@@ -494,6 +516,7 @@ export const P_AddOperationsBundle = (deps: { log: Logger }) =>
           bundleEntity.id,
           userSession.accountId,
           channelClient,
+          channelContractId,
           deps,
         );
 

src/core/service/bundle/bundle.errors.ts

@@ -7,7 +7,8 @@ export enum BUNDLE_ERROR_CODES {
   INSUFFICIENT_UTXOS = "BND_004",
   UTXO_NOT_FOUND = "BND_005",
   SPEND_OPERATION_NOT_SIGNED = "BND_006",
-  NO_OPERATIONS_PROVIDED = "BND_007",
+  // BND_007 (NO_OPERATIONS_PROVIDED) removed — unreachable behind the request
+  // schema's operationsMLXDR.min(1) validation (#13).
   BUNDLE_NOT_FOUND = "BND_008",
   BUNDLE_ACCESS_FORBIDDEN = "BND_009",
   TOO_MANY_OPERATIONS = "BND_010",
@@ -16,6 +17,7 @@ export enum BUNDLE_ERROR_CODES {
   PP_NOT_FOUND = "BND_013",
   PP_NOT_MEMBER_OF_CHANNEL = "BND_014",
   CHANNEL_DISABLED = "BND_015",
+  CHANNEL_RPC_UNAVAILABLE = "BND_016",
 }
 
 const source = "@service/bundle";
@@ -116,6 +118,32 @@ export class CHANNEL_DISABLED extends PlatformError<{
   }
 }
 
+/**
+ * Error thrown when an on-chain read needed to admit a bundle (channel
+ * context resolution or UTXO balance lookup) does not respond within the
+ * admission timeout. Fast-fails the request instead of hanging (#2).
+ */
+export class CHANNEL_RPC_UNAVAILABLE extends PlatformError<{
+  channelContractId: string;
+}> {
+  constructor(channelContractId: string) {
+    super({
+      source,
+      code: BUNDLE_ERROR_CODES.CHANNEL_RPC_UNAVAILABLE,
+      message: "Channel network is temporarily unavailable",
+      details:
+        `An on-chain read for channel '${channelContractId}' did not respond in time.`,
+      api: {
+        status: 503,
+        message: "Channel network is temporarily unavailable",
+        details:
+          "The network could not be reached to process your bundle right now. Please try again shortly.",
+      },
+      meta: { channelContractId },
+    });
+  }
+}
+
 /**
  * Error thrown when the submitter has no APPROVED entity record.
  */
@@ -229,26 +257,6 @@ export class BUNDLE_ALREADY_EXISTS extends PlatformError<{ bundleId: string }> {
   }
 }
 
-/**
- * Error thrown when no operations are provided in the bundle
- */
-export class NO_OPERATIONS_PROVIDED extends PlatformError {
-  constructor() {
-    super({
-      source,
-      code: BUNDLE_ERROR_CODES.NO_OPERATIONS_PROVIDED,
-      message: "No operations provided",
-      details: "The operations bundle must contain at least one operation.",
-      api: {
-        status: 400,
-        message: "No operations provided",
-        details:
-          "The request must include at least one operation in the operations bundle.",
-      },
-    });
-  }
-}
-
 /**
  * Error thrown when a spend operation is not signed by the UTXO owner
  */

src/core/service/event-watcher/channel-convergence.ts

@@ -1,4 +1,5 @@
 import type { ChannelRegistry } from "./channel-registry.ts";
+import type { Logger } from "@/utils/logger/index.ts";
 
 /**
  * Subset of a council's public state (`GET /api/v1/public/council`) the provider
@@ -17,17 +18,31 @@ export interface CouncilConfigData {
 export async function fetchCouncilConfig(
   councilUrl: string,
   channelAuthId: string,
+  deps?: { log?: Logger },
 ): Promise<CouncilConfigData | null> {
   try {
     const res = await fetch(
       `${councilUrl}/api/v1/public/council?councilId=${
         encodeURIComponent(channelAuthId)
       }`,
     );
-    if (!res.ok) return null;
+    if (!res.ok) {
+      // Best-effort still, but no longer silent (#10): a persistently failing
+      // council query degrades convergence and must be visible.
+      deps?.log?.error(
+        new Error(`council config query returned ${res.status}`),
+        "fetchCouncilConfig non-OK response",
+        { councilUrl, channelAuthId, status: res.status },
+      );
+      return null;
+    }
     const { data } = await res.json();
     return data as CouncilConfigData;
-  } catch {
+  } catch (error) {
+    deps?.log?.error(error, "fetchCouncilConfig failed", {
+      councilUrl,
+      channelAuthId,
+    });
     return null;
   }
 }

src/core/service/event-watcher/event-watcher.process.ts

@@ -2,7 +2,7 @@ import type { Logger } from "@/utils/logger/index.ts";
 import type { Server } from "stellar-sdk/rpc";
 import { fetchChannelAuthEvents } from "./event-watcher.service.ts";
 import type { ChannelAuthEvent } from "./event-watcher.types.ts";
-import { withSpan } from "@/core/tracing.ts";
+import { currentTraceId, SpanStatusCode, withSpan } from "@/core/tracing.ts";
 import { recoverFromOutOfRetention } from "./retention.ts";
 import { resolveBootStartLedger } from "./start-ledger.ts";
 
@@ -43,6 +43,11 @@ export class EventWatcher {
   private rpc: Server;
   private startLedgerBlock: number | null;
   private log: Logger;
+  // Health signal (#10): the watcher runs in the background with no request to
+  // surface failures on, so its last-success / last-error is tracked here and
+  // exposed via getHealth() for the /health endpoint.
+  private lastPollOkAt: number | null = null;
+  private lastPollError: { at: number; message: string } | null = null;
 
   constructor(
     config: { contractIds: string[]; intervalMs?: number },
@@ -55,6 +60,28 @@ export class EventWatcher {
     this.log = deps.log.scope("EventWatcher");
   }
 
+  /**
+   * Background-poll health for the /health endpoint. `healthy` is false once a
+   * poll has errored more recently than the last success (or never succeeded
+   * after an error).
+   */
+  getHealth(): {
+    running: boolean;
+    lastPollOkAt: number | null;
+    lastPollError: { at: number; message: string } | null;
+    healthy: boolean;
+  } {
+    const healthy = this.lastPollError === null ||
+      (this.lastPollOkAt !== null &&
+        this.lastPollOkAt >= this.lastPollError.at);
+    return {
+      running: this.isRunning,
+      lastPollOkAt: this.lastPollOkAt,
+      lastPollError: this.lastPollError,
+      healthy,
+    };
+  }
+
   /**
    * Add a channel-auth contract to the watched set (e.g. when a PP joins a new
    * council). Idempotent; the next poll picks it up — no new watcher is spun up.
@@ -206,6 +233,7 @@ export class EventWatcher {
         // Advance cursor past the latest ledger we've seen (in memory only —
         // there is no durable cursor; converge-by-query is the recovery path).
         this.lastLedger = latestLedger + 1;
+        this.lastPollOkAt = Date.now();
       } catch (error) {
         span.addEvent("poll_error", {
           "error.message": error instanceof Error
@@ -233,7 +261,23 @@ export class EventWatcher {
           return;
         }
 
-        this.log.error(error, "EventWatcher poll error");
+        // Surface the failure (#10): the swallowing catch previously left the
+        // span un-errored and no health signal. Mark it ERROR, record the
+        // exception, and update the health state so /health can report it.
+        this.lastPollError = {
+          at: Date.now(),
+          message: error instanceof Error ? error.message : String(error),
+        };
+        span.setStatus({
+          code: SpanStatusCode.ERROR,
+          message: this.lastPollError.message,
+        });
+        span.recordException(
+          error instanceof Error ? error : new Error(String(error)),
+        );
+        this.log.error(error, "EventWatcher poll error", {
+          traceId: currentTraceId(),
+        });
       }
     });
   }