From 8af594677df4fcb1c8a83472d06565b00e1abac0 Mon Sep 17 00:00:00 2001 From: D4ryl00 Date: Thu, 25 Jun 2026 00:19:46 +0200 Subject: [PATCH 1/2] test(scenario): assert offline peer catches up after reconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an end-to-end test on the full protocol service for the catch-up path that bare orbit-db-over-mocknet tests cannot cover: a peer goes offline, misses messages, reconnects, and is expected to receive every missed message via the orbit-db head exchange — with no application-level retry. mocknet does not run the discovery poll + BackoffConnector that re-dials a dropped peer on a real network (a probe confirms a peer never reconnects on its own after the link is restored), so the test re-establishes the connection with a single ConnectPeers, modelling the one reconnect discovery would produce. The reconnect then re-fires the head exchange, which backfills the whole log; the full-service gossipsub machinery converges the join without any retry loop. Ran 30x under -race without a failure. Signed-off-by: D4ryl00 --- replication_convergence_test.go | 128 ++++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 replication_convergence_test.go diff --git a/replication_convergence_test.go b/replication_convergence_test.go new file mode 100644 index 00000000..a5354cb2 --- /dev/null +++ b/replication_convergence_test.go @@ -0,0 +1,128 @@ +package weshnet_test + +import ( + "context" + "fmt" + "testing" + "time" + + mocknet "github.com/libp2p/go-libp2p/p2p/net/mock" + "github.com/stretchr/testify/require" + + weshnet "berty.tech/weshnet/v2" + "berty.tech/weshnet/v2/pkg/protocoltypes" + "berty.tech/weshnet/v2/pkg/testutil" +) + +// TestReplicationConvergence_OfflinePeerCatchesUp asserts that a peer which +// goes offline, misses messages and then reconnects catches up on every missed +// message in a single orbit-db head exchange — with no application-level retry. +// +// It runs the full weshnet protocol service (NewTestingProtocolWithMockedPeers). +// Note the one mocknet-imposed compromise: mocknet does not run the discovery +// poll + BackoffConnector that re-dials a dropped peer on a real network (a +// probe confirms a peer never reconnects on its own after the link is +// restored), so the test re-establishes the connection itself with a single +// ConnectPeers — modelling the one network reconnect that discovery would +// produce. Everything after that is the real stack: the reconnect re-fires the +// head exchange, which backfills the whole log. No retry loop is used. +func TestScenario_ReplicationConvergenceOfflinePeerCatchesUp(t *testing.T) { + testutil.FilterStabilityAndSpeed(t, testutil.Flappy, testutil.Slow) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + logger, cleanupLog := testutil.Logger(t) + defer cleanupLog() + + mn := mocknet.New() + defer mn.Close() + + opts := weshnet.TestingOpts{ + Mocknet: mn, + Logger: logger, + ConnectFunc: weshnet.ConnectAll, + } + tps, cleanupTps := weshnet.NewTestingProtocolWithMockedPeers(ctx, t, &opts, nil, 2) + defer cleanupTps() + + alice, bob := tps[0], tps[1] + aliceID, bobID := alice.Opts.Host.ID(), bob.Opts.Host.ID() + + // Create, join and activate the group on both peers. + group := weshnet.CreateMultiMemberGroupInstance(ctx, t, alice, bob) + groupPK := group.PublicKey + + // Baseline: bob receives a message from alice while connected, proving the + // group is live before we partition them. + require.NoError(t, sendMessages(ctx, alice, groupPK, []string{"online-1"})) + waitForMessages(ctx, t, bob, groupPK, []string{"online-1"}, time.Second*20) + + // Take bob offline: sever the mocknet link (so no re-dial can succeed) and + // close the existing connection. + require.NoError(t, mn.UnlinkPeers(aliceID, bobID)) + require.NoError(t, mn.DisconnectPeers(aliceID, bobID)) + + // Alice keeps sending while bob is offline. + const offlineCount = 20 + offline := make([]string, offlineCount) + for i := range offline { + offline[i] = fmt.Sprintf("offline-%d", i) + } + require.NoError(t, sendMessages(ctx, alice, groupPK, offline)) + + // Bring bob back online: restore the link and re-establish the connection + // once. mocknet won't re-dial on its own, so this single ConnectPeers stands + // in for the discovery-driven reconnect; we do not retry it, and we never + // touch orbit-db. The reconnect re-fires the head exchange on its own. + _, err := mn.LinkPeers(aliceID, bobID) + require.NoError(t, err) + _, err = mn.ConnectPeers(aliceID, bobID) + require.NoError(t, err) + + // Assert bob catches up on every message it missed, with no manual help + // beyond that single reconnect and no head-exchange retry. + waitForMessages(ctx, t, bob, groupPK, offline, time.Second*60) +} + +func sendMessages(ctx context.Context, tp *weshnet.TestingProtocol, groupPK []byte, messages []string) error { + for _, m := range messages { + if _, err := tp.Client.AppMessageSend(ctx, &protocoltypes.AppMessageSend_Request{ + GroupPk: groupPK, + Payload: []byte(m), + }); err != nil { + return err + } + } + return nil +} + +// waitForMessages subscribes to the group and blocks until every wanted payload +// has been observed or the timeout fires. GroupMessageList replays the store +// and then tails new events, so messages backfilled after the subscription +// starts are streamed too. +func waitForMessages(ctx context.Context, t *testing.T, tp *weshnet.TestingProtocol, groupPK []byte, want []string, timeout time.Duration) { + t.Helper() + + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + remaining := map[string]struct{}{} + for _, m := range want { + remaining[m] = struct{}{} + } + + sub, err := tp.Client.GroupMessageList(ctx, &protocoltypes.GroupMessageList_Request{ + GroupPk: groupPK, + }) + require.NoError(t, err) + + for len(remaining) > 0 { + res, err := sub.Recv() + if err != nil { + require.NoError(t, err, "peer did not catch up, still missing %d/%d message(s)", len(remaining), len(want)) + return + } + delete(remaining, string(res.Message)) + } +} From b480f00413999d1862a05bdfcd09bdbc59bcf340 Mon Sep 17 00:00:00 2001 From: D4ryl00 Date: Thu, 25 Jun 2026 04:31:58 +0200 Subject: [PATCH 2/2] test(scenario): note the single reconnect is low-probability, not immune The catch-up relies on a single reconnect not tripping a go-libp2p-pubsub connect/disconnect race that can drop a peer's subscriptions; this isolated clean reconnect rarely hits it, but it is not immunity (the busier bertyreplication topology hits it ~1/100). Signed-off-by: D4ryl00 --- replication_convergence_test.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/replication_convergence_test.go b/replication_convergence_test.go index a5354cb2..684da009 100644 --- a/replication_convergence_test.go +++ b/replication_convergence_test.go @@ -26,6 +26,13 @@ import ( // ConnectPeers — modelling the one network reconnect that discovery would // produce. Everything after that is the real stack: the reconnect re-fires the // head exchange, which backfills the whole log. No retry loop is used. +// +// A single reconnect is reliable here because this isolated, clean reconnect +// rarely trips the go-libp2p-pubsub connect/disconnect race that can drop a +// peer's subscriptions on reconnect (the reconnect's new-peer event is deduped +// against the not-yet-removed stale peer entry). It is low-probability, not +// immune: the busier bertyreplication topology hits it ~1/100 and needs a +// reconnect retry loop. func TestScenario_ReplicationConvergenceOfflinePeerCatchesUp(t *testing.T) { testutil.FilterStabilityAndSpeed(t, testutil.Flappy, testutil.Slow)