core/parsigdb: add support for DutySyncMessage to parsigdb (#1219)

Adds support for `DutySyncMessage` to parsigdb.

category: feature
ticket: #1192

core/parsigdb/memory.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"sync"
 
+	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
+
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/log"
 	"github.com/obolnetwork/charon/app/z"
@@ -101,15 +103,18 @@ func (db *MemDB) StoreExternal(ctx context.Context, duty core.Duty, signedSet co
 			z.Any("pubkey", pubkey))
 
 		// Call the threshSubs (which includes SigAgg component) if sufficient signatures have been received.
-		if len(sigs) != db.threshold {
+		out, ok, err := calculateOutput(duty, sigs, db.threshold)
+		if err != nil {
+			return err
+		} else if !ok {
 			continue
 		}
 
 		for _, sub := range db.threshSubs {
 			// Clone before calling each subscriber.
 			var clones []core.ParSignedData
-			for _, sig := range sigs {
-				clone, err := sig.Clone()
+			for _, psig := range out {
+				clone, err := psig.Clone()
 				if err != nil {
 					return err
 				}
@@ -125,6 +130,31 @@ func (db *MemDB) StoreExternal(ctx context.Context, duty core.Duty, signedSet co
 	return nil
 }
 
+// calculateOutput returns partial signatures and whether threshold no. of partial signatures is obtained or not.
+func calculateOutput(duty core.Duty, sigs []core.ParSignedData, threshold int) ([]core.ParSignedData, bool, error) {
+	if duty.Type != core.DutySyncMessage {
+		return sigs, len(sigs) == threshold, nil
+	}
+
+	sigsByRoot := make(map[eth2p0.Root][]core.ParSignedData)
+	for _, sig := range sigs {
+		msg, ok := sig.SignedData.(core.SignedSyncMessage)
+		if !ok {
+			return nil, false, errors.New("invalid sync message")
+		}
+
+		sigsByRoot[msg.BeaconBlockRoot] = append(sigsByRoot[msg.BeaconBlockRoot], sig)
+	}
+
+	for _, psigs := range sigsByRoot {
+		if len(psigs) == threshold {
+			return psigs, true, nil
+		}
+	}
+
+	return nil, false, nil
+}
+
 // store returns true if the value was added to the list of signatures at the provided key
 // and returns a copy of the resulting list.
 func (db *MemDB) store(k key, value core.ParSignedData) ([]core.ParSignedData, bool, error) {

core/parsigdb/memory_internal_test.go

@@ -0,0 +1,64 @@
+// Copyright © 2022 Obol Labs Inc.
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of  MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package parsigdb
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/attestantio/go-eth2-client/spec/altair"
+	"github.com/stretchr/testify/require"
+
+	"github.com/obolnetwork/charon/core"
+	"github.com/obolnetwork/charon/testutil"
+)
+
+func TestShouldOutput(t *testing.T) {
+	const (
+		n  = 4
+		th = 3
+	)
+
+	duty := core.NewSyncMessageDuty(123)
+	root1 := testutil.RandomRoot()
+	root2 := testutil.RandomRoot()
+
+	var (
+		data []core.ParSignedData
+		msgs []*altair.SyncCommitteeMessage
+	)
+
+	for i := 0; i < n-1; i++ {
+		msg := testutil.RandomSyncCommitteeMessage()
+		msg.BeaconBlockRoot = root1
+		msgs = append(msgs, msg)
+		data = append(data, core.NewPartialSignedSyncMessage(msg, i))
+	}
+
+	out, ok, err := calculateOutput(duty, data, th)
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.True(t, reflect.DeepEqual(out, data[:n-1]))
+
+	for i := 0; i < n/2; i++ {
+		msgs[i].BeaconBlockRoot = root2
+		data[i] = core.NewPartialSignedSyncMessage(msgs[i], i)
+	}
+
+	_, ok, err = calculateOutput(duty, data, th)
+	require.NoError(t, err)
+	require.False(t, ok)
+}

core/signeddata.go

@@ -805,6 +805,19 @@ func (s *SignedAggregateAndProof) UnmarshalJSON(input []byte) error {
 	return s.SignedAggregateAndProof.UnmarshalJSON(input)
 }
 
+// NewSignedSyncMessage is a convenience function which returns new signed SignedSyncMessage.
+func NewSignedSyncMessage(data *altair.SyncCommitteeMessage) SignedSyncMessage {
+	return SignedSyncMessage{SyncCommitteeMessage: *data}
+}
+
+// NewPartialSignedSyncMessage is a convenience function which returns a new partially signed SignedSyncMessage.
+func NewPartialSignedSyncMessage(data *altair.SyncCommitteeMessage, shareIdx int) ParSignedData {
+	return ParSignedData{
+		SignedData: NewSignedSyncMessage(data),
+		ShareIdx:   shareIdx,
+	}
+}
+
 // SignedSyncMessage wraps altair.SyncCommitteeMessage and implements SignedData.
 type SignedSyncMessage struct {
 	altair.SyncCommitteeMessage

core/types.go

@@ -211,6 +211,19 @@ func NewAggregatorDuty(slot int64) Duty {
 	}
 }
 
+// NewSyncMessageDuty returns a new sync message duty. It is a convenience function that is
+// slightly more readable and concise than the struct literal equivalent:
+//
+//	core.Duty{Slot: slot, Type: core.DutySyncMessage}
+//	vs
+//	core.NewSyncMessageDuty(slot)
+func NewSyncMessageDuty(slot int64) Duty {
+	return Duty{
+		Slot: slot,
+		Type: DutySyncMessage,
+	}
+}
+
 const (
 	pkLen  = 98 // "0x" + hex.Encode([48]byte) = 2+2*48
 	sigLen = 96

testutil/random.go

@@ -404,6 +404,15 @@ func RandomSyncCommitteeContribution() *altair.SyncCommitteeContribution {
 	}
 }
 
+func RandomSyncCommitteeMessage() *altair.SyncCommitteeMessage {
+	return &altair.SyncCommitteeMessage{
+		Slot:            RandomSlot(),
+		BeaconBlockRoot: RandomRoot(),
+		ValidatorIndex:  RandomVIdx(),
+		Signature:       RandomEth2Signature(),
+	}
+}
+
 func RandomSyncAggregate(t *testing.T) *altair.SyncAggregate {
 	t.Helper()