diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go index 0e4ead7e0..a3e9a18ec 100644 --- a/dkg/pedersen/board.go +++ b/dkg/pedersen/board.go @@ -72,15 +72,18 @@ var ( // In the future Kyber fork we will address this and fix all logging as well. func NewBoard(ctx context.Context, host host.Host, config *Config, bcastComp *bcast.Component) *Board { board := &Board{ - logCtx: log.WithTopic(ctx, "pedersen"), - host: host, - sender: new(p2p.Sender), - config: config, - bcastComp: bcastComp, - dedup: newBundleDedup(), - dealCh: make(chan kdkg.DealBundle), - responseCh: make(chan kdkg.ResponseBundle), - justificationCh: make(chan kdkg.JustificationBundle), + logCtx: log.WithTopic(ctx, "pedersen"), + host: host, + sender: new(p2p.Sender), + config: config, + bcastComp: bcastComp, + dedup: newBundleDedup(), + // Buffer bundle channels to cluster size: peers ahead in the ceremony send + // bundles before this node's kyber protocol reads them, and each node sends + // at most one bundle of each type per protocol run. + dealCh: make(chan kdkg.DealBundle, config.Nodes()), + responseCh: make(chan kdkg.ResponseBundle, config.Nodes()), + justificationCh: make(chan kdkg.JustificationBundle, config.Nodes()), dealOutCh: make(chan kdkg.DealBundle), responseOutCh: make(chan kdkg.ResponseBundle), justificationOutCh: make(chan kdkg.JustificationBundle), @@ -334,6 +337,8 @@ func (b *Board) handleDealBundleMessage(ctx context.Context, peerID peer.ID, msg select { case b.dealCh <- bundle: case <-ctx.Done(): + // Forget the dropped bundle so a redelivery is not refused as a duplicate. + b.dedup.forget(dealBundleMsg, protoBundle.GetSignature()) log.Error(b.logCtx, "Dropping deal bundle, context done", nil, z.Str("from", peerID.String())) } @@ -362,6 +367,8 @@ func (b *Board) handleResponseBundleMessage(ctx context.Context, peerID peer.ID, select { case b.responseCh <- bundle: case <-ctx.Done(): + // Forget the dropped bundle so a redelivery is not refused as a duplicate. + b.dedup.forget(respBundleMsg, protoBundle.GetSignature()) log.Error(b.logCtx, "Dropping response bundle, context done", nil, z.Str("from", peerID.String())) } @@ -388,6 +395,8 @@ func (b *Board) handleJustificationBundleMessage(ctx context.Context, peerID pee select { case b.justificationCh <- bundle: case <-ctx.Done(): + // Forget the dropped bundle so a redelivery is not refused as a duplicate. + b.dedup.forget(justBundleMsg, protoBundle.GetSignature()) log.Error(b.logCtx, "Dropping justification bundle, context done", nil, z.Str("from", peerID.String())) } @@ -444,6 +453,16 @@ func (d *bundleDedup) isDuplicate(msgType string, signature []byte) bool { return false } +// forget removes a bundle from the seen set, allowing it to be delivered again. +func (d *bundleDedup) forget(msgType string, signature []byte) { + h := sha256.Sum256(append([]byte(msgType), signature...)) + + d.mu.Lock() + defer d.mu.Unlock() + + delete(d.seen, h) +} + // forwardBundles forwards bundles from in to out until ctx is canceled, then closes out. // Kyber FastSync protocol goroutines exit when a board channel closes, so this stops // them without modifying the kyber library. diff --git a/dkg/pedersen/board_internal_test.go b/dkg/pedersen/board_internal_test.go new file mode 100644 index 000000000..83f80c426 --- /dev/null +++ b/dkg/pedersen/board_internal_test.go @@ -0,0 +1,238 @@ +// Copyright © 2022-2026 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1 + +package pedersen + +import ( + "context" + "testing" + "time" + + "github.com/drand/kyber" + kdkg "github.com/drand/kyber/share/dkg" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/stretchr/testify/require" + + "github.com/obolnetwork/charon/cluster" + "github.com/obolnetwork/charon/testutil" +) + +// bundleKind abstracts delivering a bundle of one type via its p2p handler and +// receiving it from the corresponding incoming board channel. +type bundleKind struct { + name string + deliver func(ctx context.Context, t *testing.T, board *Board, from peer.ID, idx uint32, session, sig []byte) + receive func(t *testing.T, board *Board) (uint32, bool) +} + +func bundleKinds() []bundleKind { + return []bundleKind{ + { + name: "deal", + deliver: func(ctx context.Context, t *testing.T, board *Board, from peer.ID, idx uint32, session, sig []byte) { + t.Helper() + + protoBundle, err := DealBundleToProto(kdkg.DealBundle{ + DealerIndex: idx, + Deals: []kdkg.Deal{ + { + ShareIndex: 1, + EncryptedShare: []byte{1, 2, 3}, + }, + }, + Public: []kyber.Point{RandomPoint(t)}, + SessionID: session, + Signature: sig, + }) + require.NoError(t, err) + + _, _, err = board.handleDealBundleMessage(ctx, from, protoBundle) + require.NoError(t, err) + }, + receive: func(t *testing.T, board *Board) (uint32, bool) { + t.Helper() + + select { + case bundle := <-board.IncomingDeal(): + return bundle.DealerIndex, true + case <-time.After(time.Second): + return 0, false + } + }, + }, + { + name: "response", + deliver: func(ctx context.Context, t *testing.T, board *Board, from peer.ID, idx uint32, session, sig []byte) { + t.Helper() + + protoBundle, err := ResponseBundleToProto(kdkg.ResponseBundle{ + ShareIndex: idx, + Responses: []kdkg.Response{ + { + DealerIndex: 1, + Status: true, + }, + }, + SessionID: session, + Signature: sig, + }) + require.NoError(t, err) + + _, _, err = board.handleResponseBundleMessage(ctx, from, protoBundle) + require.NoError(t, err) + }, + receive: func(t *testing.T, board *Board) (uint32, bool) { + t.Helper() + + select { + case bundle := <-board.IncomingResponse(): + return bundle.ShareIndex, true + case <-time.After(time.Second): + return 0, false + } + }, + }, + { + name: "justification", + deliver: func(ctx context.Context, t *testing.T, board *Board, from peer.ID, idx uint32, session, sig []byte) { + t.Helper() + + protoBundle, err := JustificationBundleToProto(kdkg.JustificationBundle{ + DealerIndex: idx, + Justifications: []kdkg.Justification{ + { + ShareIndex: 1, + Share: RandomScalar(t), + }, + }, + SessionID: session, + Signature: sig, + }) + require.NoError(t, err) + + _, _, err = board.handleJustificationBundleMessage(ctx, from, protoBundle) + require.NoError(t, err) + }, + receive: func(t *testing.T, board *Board) (uint32, bool) { + t.Helper() + + select { + case bundle := <-board.IncomingJustification(): + return bundle.DealerIndex, true + case <-time.After(time.Second): + return 0, false + } + }, + }, + } +} + +// TestBoardBuffersEarlyBundles verifies that bundles arriving before the kyber +// protocol starts reading the incoming board channels are buffered and delivered +// later, instead of being dropped when the p2p request context expires. This +// happens when peers start the DKG a few seconds earlier than this node. +func TestBoardBuffersEarlyBundles(t *testing.T) { + const ( + numNodes = 4 + threshold = 3 + ) + + for _, kind := range bundleKinds() { + t.Run(kind.name, func(t *testing.T) { + board, nodes, session := newTestBoard(t, numNodes, threshold) + + // Deliver bundles from all other peers before anything reads the + // incoming channel. Each handler call gets a short-lived context, + // like an expiring p2p request context. + for i := 1; i < numNodes; i++ { + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + kind.deliver(ctx, t, board, nodes[i].NodeHost.ID(), uint32(i), session, []byte{byte(i)}) + cancel() + } + + // All early bundles must be delivered once the protocol starts reading. + received := make(map[uint32]bool) + + for i := 1; i < numNodes; i++ { + idx, ok := kind.receive(t, board) + if !ok { + require.Failf(t, "early bundle dropped", "received %d of %d bundles", len(received), numNodes-1) + } + + received[idx] = true + } + + require.Len(t, received, numNodes-1) + }) + } +} + +// TestBoardAcceptsRedeliveryAfterDrop verifies that a bundle dropped because the +// request context expired is not recorded as a duplicate, so a redelivery of the +// same bundle is accepted. +func TestBoardAcceptsRedeliveryAfterDrop(t *testing.T) { + const ( + numNodes = 4 + threshold = 3 + ) + + for _, kind := range bundleKinds() { + t.Run(kind.name, func(t *testing.T) { + board, nodes, session := newTestBoard(t, numNodes, threshold) + from := nodes[1].NodeHost.ID() + + // Fill the channel buffer plus the forwarder's in-flight slot so the + // next delivery blocks and gets dropped. + total := numNodes + 1 + for i := range total { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + kind.deliver(ctx, t, board, from, uint32(i), session, []byte{byte(i)}) + cancel() + } + + // This bundle is dropped since nothing is reading yet and the buffer is full. + droppedIdx := uint32(total) + droppedSig := []byte{byte(total)} + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + kind.deliver(ctx, t, board, from, droppedIdx, session, droppedSig) + cancel() + + // Drain the buffered bundles. + for range total { + _, ok := kind.receive(t, board) + require.True(t, ok, "buffered bundle not delivered") + } + + // Redeliver the dropped bundle. It must not be refused as a duplicate. + ctx, cancel = context.WithTimeout(t.Context(), time.Second) + defer cancel() + + kind.deliver(ctx, t, board, from, droppedIdx, session, droppedSig) + + idx, ok := kind.receive(t, board) + require.True(t, ok, "redelivered bundle refused as duplicate") + require.Equal(t, droppedIdx, idx) + }) + } +} + +func newTestBoard(t *testing.T, numNodes, threshold int) (*Board, []*TestNode, []byte) { + t.Helper() + + var ( + peers []peer.ID + peerMap = make(map[peer.ID]cluster.NodeIdx) + session = testutil.RandomArray32() + ) + + nodes := make([]*TestNode, numNodes) + for i := range numNodes { + nodes[i] = NewTestNode(t, i) + peerMap[nodes[i].NodeHost.ID()] = nodes[i].NodeIdx + peers = append(peers, nodes[i].NodeHost.ID()) + } + + nodes[0].InitBoard(t, threshold, peers, peerMap, session[:]) + + return nodes[0].Board, nodes, session[:] +}