Fix decentralized clients connection issue by ahzero7d1 · Pull Request #1110 · epfml/disco

ahzero7d1 · 2026-04-15T14:01:39Z

Solves issue: #1109

This pull request tackles decentralized learning peer connection stability and synchronization issues. We are mainly solving 3 main issues.

Connection Ready Check and Signaling Weight Sharing

ConnectionsRetry, StartWeightSharing messages are newly added to prevent faster peers proceed to weight sharing while slower peers are still establishing connections. Detailed process is as follows:

After completing peer connections, each client sends a ConnectionsReady message to the server.
The server counts the number of received ConnectionsReady messages.
Once the number of ConnectionsReady messages matches the number of expected participants for the round, the server sends a StartWeightSharing message to all round participants.
Clients only start sharing model updates after receiving StartWeightSharing.

Connection Retries and Failed Client Disconnection

RetryPeerConnections, ConnectionFail messages are newly added. In addition, webapp shows error message when a client is disconnected after repetitive failure. maxConnectionRetry parameter is added to trainingInformation, and added to the webapp so that users can specify the value during task creation.
Connection failure handling process is as follows:

If the number of retries is still below maxConnectionRetry, the server sends a RetryPeerConnection message to all peers in the current round.
When clients receive RetryPeerConnection, they clean up their peer pool and aggregator nodes, then rerun the peer connection phase.
If the connection setup still fails after more than maxConnectionRetry attempts, the server removes the failed peers from the round peer list.
The server sends a ConnectionFail message to the failed peers.
When a client receives ConnectionFail, it disconnects from the server.
The remaining clients receive RetryPeerConnection and retry the peer connection phase without the failed peers.

Model Syncing for Participants Joining in the Middle of Training

Model syncing between peers is implemented for participants who are joining in the middle of training. ModelSyncRequest, SignalModelProvider, SignalNewPeer, SharedModel messages are newly added. In addition, ModelWeightAccess interface was added in disco.ts so that client object can adjust model weights when it receives the latest model from the provider peer.
Model syncing process is as follows:

When a new participant joins in the middle of training, the server marks the participant as having joined mid-training in NewDecentralizedNodeInfo.
After receiving NewDecentralizedNodeInfo, the new client sets a local flag indicating that it needs model syncing.
When training begins, if this flag is set, the new client sends a ModelSyncRequest message to the server.
After receiving ModelSyncRequest, the server sends messages as step 5 and 6, using selected model provider information from previous training round.
The server sends SignalModelProvider to the new participant with information about the provider peer.
The server sends SignalNewPeer to the provider peer with information about the newly joined peer.
Using this signaling information, the new participant and provider peer establish a peer connection.
The provider waits until the current aggregation round has finished, then sends the latest model to the new participant using a SharedModel message.
The new participant receives the model and updates its local model weights.
After syncing, the new participant can join subsequent decentralized training rounds.

…ing training

JulienVig

Great work! I have a few refactoring comments but nothing blocking. I'll approve the merge once the next demo is passed :)

JulienVig · 2026-06-04T10:43:35Z

+      // If this participant was a latest model provider node,
+      // replace the provider node to another node
+      if (this.#providerNode === peerId) {
+        this.#providerNode = this.connections.keySeq().first()


Maybe ensure that the new providerNode is not already in #synchingNodes here

JulienVig · 2026-06-04T11:04:59Z

+      const syncConnection = await this.#pool.getPeers(
+        Set([providerInfo.providerNode]),
+        this.server,
+        ()=>{}
+      )


Calling getPeers adds the providerNode to the peer pool (this line) which is then never removed and unused in aggregation either.
I don't think that's a big deal as I assume this model sync won't happen that often but worth adding a comment to help debug if it becomes problematic

JulienVig · 2026-06-04T11:13:04Z

+      } else if (msg.type === type.RetryPeerConnections){
+        debug(`[${shortenId(this.ownId)}] retrying peer connection establishment`)
+        // clear the communication round peer pool
+        await this.#pool?.shutdown()
+        this.#pool = new PeerPool(this.ownId)
+        // clear the connections
+        this.#connections = Map()
+        this.setAggregatorNodes(Set(this.ownId))
+        continue
+      } else if (msg.type === type.ConnectionFail){
+        debug(`[${shortenId(this.ownId)}] disconnect from the server`)
+        await this.disconnect()
+        throw new Error("Client disconnected after connection failure")
+      }
+    }
    // Exchange weight updates with peers and return aggregated weights
-    return await this.exchangeWeightUpdates(weights)
+    const aggregatedWeight = await this.exchangeWeightUpdates(weights)


Is this scenario ever covered in the unit tests?

JulienVig · 2026-06-04T11:18:48Z

+// Server signals peers to reestablish peer connections
+export interface RetryPeerConnections {
+  type: type.RetryPeerConnections
+  aggregationRound: number


I don't think aggregationRound is ever used with RetryPeerConnections, is it on purpose?

JulienVig · 2026-06-04T11:24:09Z

    case type.PeersForRound:
      return 'peers' in o && Array.isArray(o.peers) && o.peers.every(isNodeID)
+    case type.SignalNewPeer:
+      return 'newNode' in o && isNodeID(o.newNode)


If I'm not mistaken SignalNewPeer is always sent by the server to a client, if so there's no need to check its properties, you can return true directly (as you're doing for SignalModelProvider)

JulienVig · 2026-06-04T13:26:01Z

+    this.#connectionRetry += 1;
+
+    // If the number of retries exceeds the threshold, exclude the failed peers from the round
+    // and retry peer connection only with the remaining peers
+    if (this.#connectionRetry >= this.task.trainingInformation.maxConnectionRetry){


With the current logic, peers are excluded after two retries if maxConnectionRetry = 3 while I would expect the logic to try 3 times and exclude on the fourth timeout.

JulienVig · 2026-06-04T13:29:32Z

+  /**
+   * Set a timeout to check peer connections establishment
+   */
+  private startTimeout(maxTime: number = 60_000): void {


I wonder if we should make this timeout a task parameter task like maxConnectionRetry, what do you think?

JulienVig · 2026-06-04T13:32:42Z

+   * Receive model from the model provider.
+   */
+  private async receiveModel(providerConn: PeerConnection): Promise<WeightsContainer>{
+    const message = await waitMessageWithTimeout(providerConn, type.SharedModel, 30_000, "Timeout while waiting for the latest model")


I think we should make this timeout value a task parameter, depending on the model size it can make sense to increase the timeout

JulienVig · 2026-06-04T13:48:34Z

+    // Set ModelWeightAccess of the client
+    this.#client.setModelWeightAccess({
+      getModelWeight: () => {
+        return new WeightsContainer(this.trainer.model.weights.weights.map(t => t.clone()));
+      },
+      setModelWeight: (weights) => {
+        this.trainer.model.weights = weights;
+      }
+    });
    // Simply propagate the training status events emitted by the client
    this.#client.on("status", (status) => this.emit("status", status));
    this.#client.on("participants", (nbParticipants) => this.emit("participants", nbParticipants));
+    this.#client.on("modelSynced", (latestWeights) => this.emit("modelSynced", latestWeights));


Maybe you can avoid this new ModelWeightAccess interface (that is only useful for decentralized and needs to be created after initialization) if you do this:

this.#client.on("modelSynced", (latestWeights) => { this.trainer.model.weights = weights; this.emit("modelSynced", latestWeights) });

And to access the model from decentralized_client.ts, you can store a copy of the model like latestModel as an attribute of decentralized client that you can set in finishRound by having the trainer add the model as argument l144 of trainer.ts:

this.#client.finishRound(networkWeights);

Let me know what you think or if my description is too vague.

JulienVig · 2026-06-04T13:49:10Z

+      const msg = await Promise.race([
+        waitMessage(this.server, type.StartWeightSharing),
+        waitMessage(this.server, type.RetryPeerConnections),
+        waitMessage(this.server, type.ConnectionFail),
+      ])


This creates two event listeners that are not resolved per round and accumulates throughout the rounds. I don't think it's a huge deal but can you add a comment/TODO to note this?

ahzero7d1 and others added 9 commits April 15, 2026 14:52

Add ConnectionsReady, StartWeightSharing messages

2900c65

Update gitignore

dd22a7d

Fix lint error

7d0e634

Add server timeout and retry handling for peer connection failures

947d2c6

Merge branch 'develop' into decentralized-connection-fix

18e777e

Implement decentralized model synchronization for clients joining dur…

b443f53

…ing training

Improve model syncing and add max retry parameter

9c7c488

Fix promise returning problem during SignalNewPeer message processing

09ced04

Specify event type

14b2a4b

JulienVig requested changes Jun 4, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix decentralized clients connection issue#1110

Fix decentralized clients connection issue#1110
ahzero7d1 wants to merge 9 commits into
developfrom
decentralized-connection-fix

ahzero7d1 commented Apr 15, 2026 •

edited

Loading

Uh oh!

JulienVig left a comment

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

JulienVig Jun 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

ahzero7d1 commented Apr 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Connection Ready Check and Signaling Weight Sharing

Connection Retries and Failed Client Disconnection

Model Syncing for Participants Joining in the Middle of Training

Uh oh!

JulienVig left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ahzero7d1 commented Apr 15, 2026 •

edited

Loading