ruvnet/ruflo · error · Error

FederationTransport: closed

Error message

FederationTransport: closed

What it means

FederationTransport.send() checks the closed flag that close() sets, and refuses any post-close request synchronously. close() also rejects every pending correlated request with this same message, so a send that raced shutdown fails too. Transport instances are single-use: once closed they cannot be reopened — build a new FederationTransport to send again.

Source

Thrown at v3/@claude-flow/swarm/src/consensus/federation-transport.ts:154

  onMessage(handler: ConsensusMessageHandler): void {
    this.handler = handler;
  }

  peers(): readonly string[] {
    return this.peerIdsFn().filter(id => id !== this.nodeId);
  }

  private stamp(msg: Omit<ConsensusMessage, 'from'>): ConsensusMessage {
    const base: Omit<ConsensusMessage, 'signature'> = {
      ...msg,
      from: this.nodeId,
      seq: this.keyPair ? ++this.seqCounter : msg.seq,
    };
    return this.keyPair ? { ...base, signature: signMessage(base, this.keyPair.privateKeyPem) } : base;
  }

  async send(to: string, msg: Omit<ConsensusMessage, 'from'>, timeoutMs?: number): Promise<ConsensusReply> {
    if (this.closed) throw new Error('FederationTransport: closed');
    const addr = this.addressOf(to);
    if (!addr) throw new Error(`FederationTransport: no address for peer ${to}`);
    const corr = randomBytes(8).toString('hex');
    const stamped = this.stamp({ ...msg, to });
    const t = timeoutMs ?? this.defaultTimeoutMs;
    return new Promise<ConsensusReply>((resolve, reject) => {
      const timer = setTimeout(() => { this.pending.delete(corr); reject(new Error(`FederationTransport: send to ${to} timed out (${t}ms)`)); }, t);
      this.pending.set(corr, { resolve, reject, timer });
      this.wire.send(addr, { type: 'consensus', payload: { corr, msg: stamped } as WireEnvelope, streamId: this.streamId })
        .catch((e) => { clearTimeout(timer); this.pending.delete(corr); reject(e instanceof Error ? e : new Error(String(e))); });
    });
  }

  async broadcast(msg: Omit<ConsensusMessage, 'from'>): Promise<void> {
    if (this.closed) throw new Error('FederationTransport: closed');
    const stamped = this.stamp(msg);
    await Promise.allSettled(this.peers().map(async (to) => {
      const addr = this.addressOf(to);

View on GitHub (pinned to fa13ee4ad6)

Solutions

1. Fix shutdown ordering: stop/quiesce consensus senders first, then transport.close().
2. Track closed state in the caller (wrap close) and skip or reconnect before send.
3. Create a new FederationTransport instance instead of reusing a closed one.
4. Catch this rejection for in-flight requests and retry once on a fresh transport.

Example fix

// before
await transport.close();
await transport.send(peer, msg); // throws: closed

// after — quiesce senders first; recreate to reuse
await consensus.shutdown(); // no more senders
await transport.close();
// to send again, build a new FederationTransport on a fresh wire

// wrap the transport so callers can see lifecycle state
let closed = false;
const origClose = transport.close.bind(transport);
transport.close = async () => { closed = true; await origClose(); };
if (closed) {
  throw new Error('transport closed; recreate it before sending');
}
await transport.send(peer, msg);

try {
  await transport.send(peer, msg);
} catch (e) {
  if (e instanceof Error && e.message === 'FederationTransport: closed') {
    // shutdown raced the send: stop sending, or rebuild the transport and retry once
  } else {
    throw e;
  }
}

When it happens

Trigger: Calling send() after awaiting close(); shutdown beginning while a request is in flight (its entry in the pending map gets rejected with this error); one subsystem closing a shared transport while the consensus layer still sends on it.

Common situations: App shutdown ordering bugs — the network/transport layer closes before consensus senders quiesce; reconnect logic closing the old transport instance while callers still hold a reference to it.

Related errors

• Connection pool is shutting down
• HTTP transport already running
• Transport "${name}" already exists
• TransportManager already running
• Stdio transport already running

AI-assisted analysis of ruvnet/ruflo@fa13ee4ad6 (2026-08-18). Data as JSON: /api/errors/e63ea257a9015619. Report an issue: GitHub.