fix: resolve issues #751-754 — reconnect backoff, threshold checks, merge events, half-open probes

src/accounts/AccountMergeDetector.ts

@@ -29,6 +29,13 @@ export class InvalidDestinationError extends Error {
   }
 }
 
+/** Payload emitted on the "merge" event. */
+export interface MergeEventPayload {
+  source: string;
+  destination: string;
+  mergedAt: Date;
+}
+
 export class AccountMergeDetector extends EventEmitter {
   private watchedAccounts = new Set<string>();
   private mergeCache = new Map<string, string>(); // source -> destination mapping
@@ -187,6 +194,12 @@ export class AccountMergeDetector extends EventEmitter {
 
     this.emit("recipient:mergeDetected", event);
 
+    this.emit("merge", {
+      source: sourceAccount,
+      destination: destinationAccount,
+      mergedAt: event.timestamp,
+    } satisfies MergeEventPayload);
+
     // Notify the client to reroute recipients
     try {
       // The client will handle rerouting via rerouteRecipient method

src/accounts/AccountSignerWeightCalculator.ts

@@ -105,6 +105,25 @@ export class AccountSignerWeightCalculator {
     this.cache.delete(accountId);
   }
 
+  /**
+   * Check whether the provided signing keys meet the required threshold
+   * for the given account. Returns `true` if sufficient, `false` otherwise.
+   *
+   * Missing signers (not on the account) contribute 0 weight.
+   *
+   * @param accountId      - The Stellar account G… address.
+   * @param signers        - List of public keys that will sign the transaction.
+   * @param threshold      - Which threshold level to check: 'low', 'medium', or 'high'.
+   */
+  async meetsThreshold(
+    accountId: string,
+    signers: string[],
+    threshold: ThresholdLevel,
+  ): Promise<boolean> {
+    const result = await this.calculateWeight(accountId, signers, threshold);
+    return result.sufficient;
+  }
+
   // --------------------------------------------------------------------------
   // Private helpers
   // --------------------------------------------------------------------------

src/horizon/HorizonStreamManager.ts

@@ -90,6 +90,7 @@ export type HorizonStreamKind = "payments" | "operations";
 export const DEFAULT_REPLAY_CUTOFF_MS = 300_000;
 export const DEFAULT_DEDUPE_BUFFER_SIZE = 256;
 export const DEFAULT_RECONNECT_DELAY_MS = 1_000;
+export const MAX_RECONNECT_DELAY_MS = 30_000;
 
 export interface HorizonStreamManagerConfig<T extends HorizonStreamRecord = HorizonStreamRecord> {
   /** Horizon server URL. Required unless `source` is supplied directly (e.g. for tests). */
@@ -108,15 +109,19 @@ export interface HorizonStreamManagerConfig<T extends HorizonStreamRecord = Hori
   dedupeBufferSize?: number;
   /** Delay before reconnecting after a stream error. Default: 1 000ms. */
   reconnectDelayMs?: number;
+  /** Maximum number of reconnect attempts before giving up. Default: Infinity (unlimited). */
+  maxReconnectAttempts?: number;
   /** Time source — exposed for deterministic tests. */
   now?: () => number;
 }
 
 /** Event map for {@link HorizonStreamManager}. */
 export interface HorizonStreamEventMap {
   "stream:reconnected": [{ accountId: string; cursor: string | null }];
+  "stream:reconnecting": [{ accountId: string; attempt: number; delayMs: number }];
   "stream:lag": [{ accountId: string; error: unknown }];
   "stream:cursor_advanced": [{ accountId: string; cursor: string }];
+  "stream:reconnect_failed": [{ accountId: string; attempts: number }];
 }
 
 function defaultSource<T extends HorizonStreamRecord>(
@@ -144,6 +149,7 @@ export class HorizonStreamManager<
   private readonly _replayCutoffMs: number;
   private readonly _dedupeBufferSize: number;
   private readonly _reconnectDelayMs: number;
+  private readonly _maxReconnectAttempts: number;
   private readonly _now: () => number;
 
   private _accountId: string | null = null;
@@ -152,6 +158,7 @@ export class HorizonStreamManager<
   private _closeStream: (() => void) | null = null;
   private _reconnectTimer: ReturnType<typeof setTimeout> | null = null;
   private _stopped = true;
+  private _reconnectAttempts = 0;
   private _seenTokens: string[] = [];
   private readonly _seenSet = new Set<string>();
 
@@ -172,6 +179,7 @@ export class HorizonStreamManager<
     this._replayCutoffMs = config.replayCutoffMs ?? DEFAULT_REPLAY_CUTOFF_MS;
     this._dedupeBufferSize = config.dedupeBufferSize ?? DEFAULT_DEDUPE_BUFFER_SIZE;
     this._reconnectDelayMs = config.reconnectDelayMs ?? DEFAULT_RECONNECT_DELAY_MS;
+    this._maxReconnectAttempts = config.maxReconnectAttempts ?? Infinity;
     this._now = config.now ?? (() => Date.now());
   }
 
@@ -181,6 +189,7 @@ export class HorizonStreamManager<
     this._accountId = accountId;
     this._handler = handler;
     this._stopped = false;
+    this._reconnectAttempts = 0;
     this._cursor = this._cursorStore.get(this._cursorKey(accountId));
     this._seenTokens = [];
     this._seenSet.clear();
@@ -225,6 +234,7 @@ export class HorizonStreamManager<
     });
 
     if (isReconnect) {
+      this._reconnectAttempts = 0;
       this.emit("stream:reconnected", { accountId: this._accountId, cursor: this._cursor });
     }
   }
@@ -270,9 +280,28 @@ export class HorizonStreamManager<
     this._closeStream?.();
     this._closeStream = null;
     this.emit("stream:lag", { accountId: this._accountId, error: event });
+
+    this._reconnectAttempts += 1;
+    if (this._reconnectAttempts > this._maxReconnectAttempts) {
+      this.emit("stream:reconnect_failed", {
+        accountId: this._accountId,
+        attempts: this._reconnectAttempts,
+      });
+      return;
+    }
+
+    const delayMs = Math.min(
+      this._reconnectDelayMs * Math.pow(2, this._reconnectAttempts - 1),
+      MAX_RECONNECT_DELAY_MS,
+    );
+    this.emit("stream:reconnecting", {
+      accountId: this._accountId,
+      attempt: this._reconnectAttempts,
+      delayMs,
+    });
     this._reconnectTimer = setTimeout(() => {
       this._reconnectTimer = null;
       this._connect(true);
-    }, this._reconnectDelayMs);
+    }, delayMs);
   }
 }

src/resilience/CircuitBreaker.ts

@@ -24,6 +24,8 @@ export interface CircuitBreakerOptions {
   openDurationMs: number;
   /** Milliseconds before an in-flight HALF_OPEN probe is treated as a failure. */
   halfOpenProbeTimeoutMs: number;
+  /** Number of concurrent probe requests allowed in HALF_OPEN state. Default: 1. */
+  halfOpenProbeCount?: number;
 }
 
 type ResolvedCircuitBreakerOptions = Required<CircuitBreakerOptions>;
@@ -33,6 +35,7 @@ const DEFAULT_OPTIONS: ResolvedCircuitBreakerOptions = {
   successThreshold: 1,
   openDurationMs: 30_000,
   halfOpenProbeTimeoutMs: 5_000,
+  halfOpenProbeCount: 1,
 };
 
 /** Structured log event emitted on every state transition. */
@@ -63,7 +66,7 @@ interface MutableState {
   successCount: number;
   lastFailureAt: number;
   lastTransitionAt: number;
-  halfOpenProbeInFlight: boolean;
+  halfOpenProbesInFlight: number;
 }
 
 export class CircuitBreaker {
@@ -81,7 +84,7 @@ export class CircuitBreaker {
       successCount: 0,
       lastFailureAt: 0,
       lastTransitionAt: now,
-      halfOpenProbeInFlight: false,
+      halfOpenProbesInFlight: 0,
     };
   }
 
@@ -105,10 +108,10 @@ export class CircuitBreaker {
     }
 
     if (this.state.state === "HALF_OPEN") {
-      if (this.state.halfOpenProbeInFlight) {
-        throw new CircuitOpenError({ state: this.state.state, reason: "probe_in_flight" });
+      if (this.state.halfOpenProbesInFlight >= this.options.halfOpenProbeCount) {
+        throw new CircuitOpenError({ state: this.state.state, reason: "probe_limit_reached" });
       }
-      this.state.halfOpenProbeInFlight = true;
+      this.state.halfOpenProbesInFlight += 1;
       try {
         const result = await this._withProbeTimeout(fn);
         this._onSuccess();
@@ -117,7 +120,7 @@ export class CircuitBreaker {
         this._onFailure();
         throw error;
       } finally {
-        this.state.halfOpenProbeInFlight = false;
+        this.state.halfOpenProbesInFlight -= 1;
       }
     }
 
@@ -136,7 +139,7 @@ export class CircuitBreaker {
     this._transition("CLOSED");
     this.state.failureCount = 0;
     this.state.successCount = 0;
-    this.state.halfOpenProbeInFlight = false;
+    this.state.halfOpenProbesInFlight = 0;
   }
 
   private _withProbeTimeout<T>(fn: () => Promise<T>): Promise<T> {