*: sync contributions per subcommittee

app/app.go

@@ -714,13 +714,21 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 	coreConsensus := consensusController.CurrentConsensus() // initially points to DefaultConsensus()
 
 	// Priority protocol always uses QBFTv2.
-	err = wirePrioritise(ctx, conf, life, p2pNode, peerIDs, lock.Threshold,
+	isync, err := wirePrioritise(ctx, conf, life, p2pNode, peerIDs, lock.Threshold,
 		sender.SendReceive, defaultConsensus, sched, p2pKey, deadlineFunc,
 		consensusController, lock.ConsensusProtocol)
 	if err != nil {
 		return err
 	}
 
+	// Gate the plural SyncContributions encoding on a threshold of peers being on a
+	// version that understands it (charon >= v1.11), re-evaluated each info-sync
+	// round. isync is nil for leader-cast (no priority protocol), leaving the
+	// fetcher on the backwards-compatible single encoding.
+	if isync != nil {
+		fetch.RegisterSyncContributionV2(isync.SyncContributionsSupported)
+	}
+
 	track, err := newTracker(ctx, life, deadlineFunc, peers, eth2Cl)
 	if err != nil {
 		return err
@@ -765,11 +773,11 @@ func wirePrioritise(ctx context.Context, conf Config, life *lifecycle.Manager, p
 	peers []peer.ID, threshold int, sendFunc p2p.SendReceiveFunc, coreCons core.Consensus,
 	sched core.Scheduler, p2pKey *k1.PrivateKey, deadlineFunc func(duty core.Duty) (time.Time, bool),
 	consensusController core.ConsensusController, clusterPreferredProtocol string,
-) error {
+) (*infosync.Component, error) {
 	cons, ok := coreCons.(*qbft.Consensus)
 	if !ok {
 		// Priority protocol not supported for leader cast.
-		return nil
+		return nil, nil //nolint:nilnil // No infosync component and no error is valid here.
 	}
 
 	// exchangeTimeout of 6 seconds (half a slot) is a good thumb suck.
@@ -779,7 +787,7 @@ func wirePrioritise(ctx context.Context, conf Config, life *lifecycle.Manager, p
 	prio, err := priority.NewComponent(ctx, p2pNode, peers, threshold,
 		sendFunc, p2p.RegisterHandler, cons, exchangeTimeout, p2pKey, deadlineFunc)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// The initial protocols order as defined by implementation is altered by:
@@ -837,7 +845,7 @@ func wirePrioritise(ctx context.Context, conf Config, life *lifecycle.Manager, p
 
 	life.RegisterStart(lifecycle.AsyncAppCtx, lifecycle.StartPeerInfo, lifecycle.HookFuncCtx(prio.Start))
 
-	return nil
+	return isync, nil
 }
 
 // newTracker creates and starts a new tracker instance.

core/aggsigdb/memory.go

@@ -43,23 +43,28 @@ type MemDB struct {
 // Store implements core.AggSigDB, see its godoc.
 func (db *MemDB) Store(ctx context.Context, duty core.Duty, set core.SignedDataSet) error {
 	for pubKey, data := range set {
-		if err := db.store(ctx, duty, pubKey, data); err != nil {
+		subcommIdx, err := core.SyncSubcommitteeIndex(duty.Type, data)
+		if err != nil {
+			return err
+		}
+
+		if err := db.store(ctx, memDBKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}, data); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-func (db *MemDB) store(ctx context.Context, duty core.Duty, pubKey core.PubKey, data core.SignedData) error {
+func (db *MemDB) store(ctx context.Context, key memDBKey, data core.SignedData) error {
 	clone, err := data.Clone() // Clone before storing.
 	if err != nil {
 		return err
 	}
 
 	response := make(chan error, 1)
 	cmd := writeCommand{
-		memDBKey: memDBKey{duty, pubKey},
+		memDBKey: key,
 		data:     clone,
 		response: response,
 	}
@@ -83,14 +88,14 @@ func (db *MemDB) store(ctx context.Context, duty core.Duty, pubKey core.PubKey,
 }
 
 // Await implements core.AggSigDB, see its godoc.
-func (db *MemDB) Await(ctx context.Context, duty core.Duty, pubKey core.PubKey) (core.SignedData, error) {
+func (db *MemDB) Await(ctx context.Context, duty core.Duty, pubKey core.PubKey, subcommIdx core.SubcommitteeIndex) (core.SignedData, error) {
 	cancel := make(chan struct{})
 	defer close(cancel)
 
 	response := make(chan core.SignedData, 1)
 
 	query := readQuery{
-		memDBKey: memDBKey{duty, pubKey},
+		memDBKey: memDBKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx},
 		response: response,
 		cancel:   cancel,
 	}
@@ -146,7 +151,7 @@ func (db *MemDB) execCommand(command writeCommand) {
 
 	_ = db.deadliner.Add(command.duty) // TODO(corver): Distinguish between no deadline supported vs already expired.
 
-	key := memDBKey{command.duty, command.pubKey}
+	key := command.memDBKey
 
 	if existing, ok := db.data[key]; ok {
 		equal, err := dataEqual(existing, command.data)
@@ -178,7 +183,7 @@ func dataEqual(x core.SignedData, y core.SignedData) (bool, error) {
 // execQuery returns true if the query was successfully executed.
 // If the requested entry is found in the DB it will return it via query.response channel.
 func (db *MemDB) execQuery(query readQuery) bool {
-	data, ok := db.data[memDBKey{query.duty, query.pubKey}]
+	data, ok := db.data[query.memDBKey]
 	if !ok {
 		return false
 	}
@@ -228,6 +233,11 @@ func cancelled(cancel <-chan struct{}) bool {
 type memDBKey struct {
 	duty   core.Duty
 	pubKey core.PubKey
+	// subcommIdx is the sync subcommittee index for sync-committee aggregator
+	// duties (DutyPrepareSyncContribution, DutySyncContribution), and 0 otherwise.
+	// A validator can occupy multiple sync subcommittees in the same slot, so it
+	// disambiguates their otherwise-colliding aggregated signatures.
+	subcommIdx core.SubcommitteeIndex
 }
 
 // writeCommand holds the data to write into the database.

core/aggsigdb/memory_internal_test.go

@@ -33,7 +33,7 @@ func TestDutyExpiration(t *testing.T) {
 	err := db.Store(ctx, duty, core.SignedDataSet{pubkey: sig})
 	require.NoError(t, err)
 
-	resp, err := db.Await(ctx, duty, pubkey)
+	resp, err := db.Await(ctx, duty, pubkey, 0)
 	require.NoError(t, err)
 	require.Equal(t, sig, resp)
 
@@ -78,7 +78,7 @@ func TestCancelledQuery(t *testing.T) {
 	errCh := make(chan error, 2)
 
 	go func() {
-		_, err := db.Await(qctx, duty, pubkey)
+		_, err := db.Await(qctx, duty, pubkey, 0)
 		errCh <- err
 
 		wg.Done()
@@ -87,7 +87,7 @@ func TestCancelledQuery(t *testing.T) {
 	require.Equal(t, 1, <-queryCount)
 
 	wg.Go(func() {
-		_, err := db.Await(qctx, duty, pubkey)
+		_, err := db.Await(qctx, duty, pubkey, 0)
 		errCh <- err
 	})
 

core/aggsigdb/memory_test.go

@@ -45,7 +45,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		err := db.Store(context.Background(), testDuty, core.SignedDataSet{testPubKey: testSignedData})
 		require.NoError(t, err)
 
-		result, err := db.Await(context.Background(), testDuty, testPubKey)
+		result, err := db.Await(context.Background(), testDuty, testPubKey, 0)
 		require.NoError(t, err)
 
 		require.EqualValues(t, testSignedData, result)
@@ -67,7 +67,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		})
 
 		go func() {
-			result, err := db.Await(context.Background(), testDuty, testPubKey)
+			result, err := db.Await(context.Background(), testDuty, testPubKey, 0)
 			resChan <- struct {
 				result core.SignedData
 				err    error
@@ -97,7 +97,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		ctx2, cancel2 := context.WithCancel(context.Background())
 
 		go func() {
-			_, err := db.Await(ctx2, testDuty, testPubKey)
+			_, err := db.Await(ctx2, testDuty, testPubKey, 0)
 			errChan <- err
 		}()
 
@@ -122,7 +122,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		ctx2, cancel2 := context.WithCancel(context.Background())
 		cancel2()
 
-		_, err := db.Await(ctx2, testDuty, testPubKey)
+		_, err := db.Await(ctx2, testDuty, testPubKey, 0)
 		require.Error(t, err)
 		require.Equal(t, "context canceled", err.Error())
 	})
@@ -143,7 +143,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		errChan := make(chan error)
 
 		go func() {
-			_, err := db.Await(context.Background(), testDuty, testPubKey)
+			_, err := db.Await(context.Background(), testDuty, testPubKey, 0)
 			errChan <- err
 		}()
 
@@ -195,7 +195,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		err = db.Store(context.Background(), testDuty, core.SignedDataSet{testPubKey: testSignedData})
 		require.NoError(t, err)
 
-		result, err := db.Await(context.Background(), testDuty, testPubKey)
+		result, err := db.Await(context.Background(), testDuty, testPubKey, 0)
 		require.NoError(t, err)
 		require.EqualValues(t, testSignedData, result)
 	})
@@ -213,7 +213,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		err := db.Store(context.Background(), testDuty, core.SignedDataSet{testPubKey: testSignedData})
 		require.NoError(t, err)
 
-		result, err := db.Await(context.Background(), testDuty, testPubKey)
+		result, err := db.Await(context.Background(), testDuty, testPubKey, 0)
 		require.NoError(t, err)
 		require.EqualValues(t, testSignedData, result)
 
@@ -224,7 +224,7 @@ func testMemDB(t *testing.T, newMemDB func(core.Deadliner) core.AggSigDB) {
 		err = db.Store(context.Background(), testDuty, core.SignedDataSet{testPubKey: testSignedData})
 		require.Equal(t, err.Error(), aggsigdb.ErrStopped.Error())
 
-		_, err = db.Await(context.Background(), testDuty, testPubKey)
+		_, err = db.Await(context.Background(), testDuty, testPubKey, 0)
 		require.Equal(t, err.Error(), aggsigdb.ErrStopped.Error())
 	})
 }

core/aggsigdb/memory_v2.go

@@ -33,15 +33,13 @@ func NewMemDBV2(deadliner core.Deadliner) *MemDBV2 {
 	}
 }
 
-func (m *MemDBV2) store(duty core.Duty, pubKey core.PubKey, data core.SignedData) error {
+func (m *MemDBV2) store(key memDBKey, data core.SignedData) error {
 	data, err := data.Clone()
 	if err != nil {
 		return err
 	}
 
-	_ = m.deadliner.Add(duty) // TODO(corver): Distinguish between no deadline supported vs already expired.
-
-	key := memDBKey{duty, pubKey}
+	_ = m.deadliner.Add(key.duty) // TODO(corver): Distinguish between no deadline supported vs already expired.
 
 	if existing, ok := m.data[key]; ok {
 		equal, err := dataEqual(existing, data)
@@ -52,7 +50,7 @@ func (m *MemDBV2) store(duty core.Duty, pubKey core.PubKey, data core.SignedData
 		}
 	} else {
 		m.data[key] = data
-		m.keysByDuty[duty] = append(m.keysByDuty[duty], key)
+		m.keysByDuty[key.duty] = append(m.keysByDuty[key.duty], key)
 	}
 
 	return nil
@@ -71,7 +69,12 @@ func (m *MemDBV2) Store(ctx context.Context, duty core.Duty, set core.SignedData
 	}
 
 	for pubKey, data := range set {
-		if err := m.store(duty, pubKey, data); err != nil {
+		subcommIdx, err := core.SyncSubcommitteeIndex(duty.Type, data)
+		if err != nil {
+			return err
+		}
+
+		if err := m.store(memDBKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}, data); err != nil {
 			return err
 		}
 	}
@@ -86,7 +89,7 @@ func (m *MemDBV2) Store(ctx context.Context, duty core.Duty, set core.SignedData
 	return nil
 }
 
-func (m *MemDBV2) Await(ctx context.Context, duty core.Duty, pubKey core.PubKey) (core.SignedData, error) {
+func (m *MemDBV2) Await(ctx context.Context, duty core.Duty, pubKey core.PubKey, subcommIdx core.SubcommitteeIndex) (core.SignedData, error) {
 	errMustLoop := errors.New("still needs loop")
 
 	query := func() (core.SignedData, error) {
@@ -99,7 +102,7 @@ func (m *MemDBV2) Await(ctx context.Context, duty core.Duty, pubKey core.PubKey)
 		case <-m.closed:
 			return nil, ErrStopped
 		default:
-			data, ok := m.data[memDBKey{duty, pubKey}]
+			data, ok := m.data[memDBKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}]
 			if !ok {
 				return nil, errMustLoop
 			}

core/aggsigdb/memory_v2_internal_test.go

@@ -29,7 +29,7 @@ func TestDutyExpirationV2(t *testing.T) {
 	err := db.Store(ctx, duty, core.SignedDataSet{pubkey: sig})
 	require.NoError(t, err)
 
-	resp, err := db.Await(ctx, duty, pubkey)
+	resp, err := db.Await(ctx, duty, pubkey, 0)
 	require.NoError(t, err)
 	require.Equal(t, sig, resp)
 
@@ -62,7 +62,7 @@ func TestCancelledQueryV2(t *testing.T) {
 	errStore := make([]error, awaitAmt)
 	for idx := range awaitAmt {
 		go func() {
-			_, err := db.Await(qctx, duty, pubkey)
+			_, err := db.Await(qctx, duty, pubkey, 0)
 			errStore[idx] = err
 
 			wg.Done()

core/dutydb/memory.go

@@ -478,18 +478,38 @@ func (db *MemDB) storeAggAttestationUnsafe(unsignedData core.UnsignedData) error
 	return nil
 }
 
-// storeSyncContributionUnsafe stores the unsigned aggregated attestation. It is unsafe since it assumes the lock is held.
+// storeSyncContributionUnsafe stores the unsigned sync committee contributions.
+// A validator can aggregate for multiple sync subcommittees in the same slot, so
+// the unsigned data is a collection of contributions (one per subcommittee), each
+// stored independently by (slot, subcommittee index, beacon block root). It is
+// unsafe since it assumes the lock is held.
 func (db *MemDB) storeSyncContributionUnsafe(unsignedData core.UnsignedData) error {
 	cloned, err := unsignedData.Clone() // Clone before storing.
 	if err != nil {
 		return err
 	}
 
-	contrib, ok := cloned.(core.SyncContribution)
-	if !ok {
-		return errors.New("invalid unsigned sync committee contribution")
+	// The unsigned data is either a plural SyncContributions (v1.11+, one per
+	// subcommittee) or, for backwards compatibility, a single SyncContribution.
+	switch contrib := cloned.(type) {
+	case core.SyncContributions:
+		for _, entry := range contrib {
+			if err := db.storeSyncContributionEntryUnsafe(entry); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	case core.SyncContribution:
+		return db.storeSyncContributionEntryUnsafe(contrib)
+	default:
+		return errors.New("invalid unsigned sync committee contributions")
 	}
+}
 
+// storeSyncContributionEntryUnsafe stores a single sync committee contribution.
+// It is unsafe since it assumes the lock is held.
+func (db *MemDB) storeSyncContributionEntryUnsafe(contrib core.SyncContribution) error {
 	contribRoot, err := contrib.HashTreeRoot()
 	if err != nil {
 		return errors.Wrap(err, "hash sync committee contribution")

core/dutydb/memory_test.go

@@ -320,7 +320,7 @@ func TestMemDBSyncContribution(t *testing.T) {
 		for range queries {
 			contrib := testutil.RandomSyncCommitteeContribution()
 			set := core.UnsignedDataSet{
-				testutil.RandomCorePubKey(t): core.NewSyncContribution(contrib),
+				testutil.RandomCorePubKey(t): core.SyncContributions{core.NewSyncContribution(contrib)},
 			}
 
 			var (
@@ -373,13 +373,13 @@ func TestMemDBSyncContribution(t *testing.T) {
 		contrib1.Slot = slot
 		contrib1.SubcommitteeIndex = subcommIdx
 		contrib1.BeaconBlockRoot = beaconBlockRoot
-		unsigned1 := core.NewSyncContribution(contrib1)
+		unsigned1 := core.SyncContributions{core.NewSyncContribution(contrib1)}
 
 		contrib2 := testutil.RandomSyncCommitteeContribution()
 		contrib2.Slot = slot
 		contrib2.SubcommitteeIndex = subcommIdx
 		contrib2.BeaconBlockRoot = beaconBlockRoot
-		unsigned2 := core.NewSyncContribution(contrib2)
+		unsigned2 := core.SyncContributions{core.NewSyncContribution(contrib2)}
 
 		// Store them.
 		err := db.Store(ctx, duty, core.UnsignedDataSet{
@@ -405,7 +405,7 @@ func TestMemDBSyncContribution(t *testing.T) {
 			testutil.RandomCorePubKey(t): testutil.RandomDenebCoreVersionedAggregateAttestation(),
 		})
 		require.Error(t, err)
-		require.ErrorContains(t, err, "invalid unsigned sync committee contribution")
+		require.ErrorContains(t, err, "invalid unsigned sync committee contributions")
 	})
 }