core/parsigdb: improve key uniqueness

app/featureset/featureset.go

@@ -45,6 +45,11 @@ const (
  // The feature gets automatically enabled when the current network is gnosis|chiado,
  // unless the user disabled this feature explicitly.
  GnosisBlockHotfix Feature = "gnosis_block_hotfix"
+
+ // HardenedParSigDB enables keying ParSigDB elements with their hash tree root
+ // on top of the core.Duty object.
+ // Since this is a breaking change, we'll gradually enable it in later releases.
+ HardenedParSigDB Feature = "hardened_par_sig_db"
 )
 
 var (
@@ -56,6 +61,7 @@ var (
  AggSigDBV2: statusAlpha,
  JSONRequests: statusAlpha,
  GnosisBlockHotfix: statusAlpha,
+ HardenedParSigDB: statusAlpha,
  // Add all features and there status here.
  }
 

core/parsigdb/memory.go

@@ -9,6 +9,7 @@ import (
  "sync"
 
  "github.com/obolnetwork/charon/app/errors"
+ "github.com/obolnetwork/charon/app/featureset"
  "github.com/obolnetwork/charon/app/log"
  "github.com/obolnetwork/charon/app/z"
  "github.com/obolnetwork/charon/core"
@@ -83,7 +84,16 @@ func (db *MemDB) StoreExternal(ctx context.Context, duty core.Duty, signedSet co
  output := make(map[core.PubKey][]core.ParSignedData)
 
  for pubkey, sig := range signedSet {
- sigs, ok, err := db.store(key{Duty: duty, PubKey: pubkey}, sig)
+ root, err := rootFromParSigDataSet(duty, sig)
+ if err != nil {
+ return errors.Wrap(err, "cannot hash tree root ParSignedData")
+ }
+
+ sigs, ok, err := db.store(key{
+ Duty: duty,
+ PubKey: pubkey,
+ Root: root,
+ }, sig)
  if err != nil {
  return err
  } else if !ok {
@@ -224,6 +234,23 @@ func getThresholdMatching(typ core.DutyType, sigs []core.ParSignedData, threshol
  return nil, false, nil
 }
 
+// rootFromParSigDataSet returns the MessageRoot of data only if the duty is supported,
+// and only if data is not a core.Signature.
+func rootFromParSigDataSet(duty core.Duty, data core.ParSignedData) ([32]byte, error) {
+ if !featureset.Enabled(featureset.HardenedParSigDB) {
+ return [32]byte{}, nil
+ }
+
+ // known limitation: sync message and contributions might have different hashes,
+ // ignore them for now.
+ // signature duties don't have a MessageRoot implementation at all, by design.
+ if duty.Type == core.DutySyncMessage || duty.Type == core.DutySyncContribution || duty.Type == core.DutySignature {
+ return [32]byte{}, nil
+ }
+
+ return data.MessageRoot()
+}
+
 func parSignedDataEqual(x, y core.ParSignedData) (bool, error) {
  xjson, err := json.Marshal(x)
  if err != nil {
@@ -240,4 +267,5 @@ func parSignedDataEqual(x, y core.ParSignedData) (bool, error) {
 type key struct {
  Duty core.Duty
  PubKey core.PubKey
+ Root [32]byte
 }

core/parsigdb/memory_internal_test.go

@@ -16,6 +16,83 @@ import (
  "github.com/obolnetwork/charon/testutil"
 )
 
+func TestRootMatching(t *testing.T) {
+ t.Run("matching roots yield threshold data", func(t *testing.T) {
+ deadliner := newTestDeadliner()
+ db := NewMemDB(2, deadliner)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ go db.Trim(ctx)
+
+ timesCalled := 0
+ db.SubscribeThreshold(func(_ context.Context, _ core.Duty, _ map[core.PubKey][]core.ParSignedData) error {
+ timesCalled++
+
+ return nil
+ })
+
+ pubkey := testutil.RandomCorePubKey(t)
+
+ duty := core.NewProposerDuty(uint64(42))
+
+ blockData := testutil.RandomDenebVersionedSignedProposal()
+
+ for i := range 2 {
+ full, err := core.NewPartialVersionedSignedProposal(blockData, i+1)
+ require.NoError(t, err)
+
+ if i == 0 {
+ require.NoError(t, db.StoreInternal(ctx, duty, core.ParSignedDataSet{
+ pubkey: full,
+ }))
+ } else {
+ require.NoError(t, db.StoreExternal(ctx, duty, core.ParSignedDataSet{
+ pubkey: full,
+ }))
+ }
+ }
+
+ require.Equal(t, 1, timesCalled)
+ })
+
+ t.Run("non-matching roots yield no threshold data", func(t *testing.T) {
+ deadliner := newTestDeadliner()
+ db := NewMemDB(2, deadliner)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ go db.Trim(ctx)
+
+ timesCalled := 0
+ db.SubscribeThreshold(func(_ context.Context, _ core.Duty, _ map[core.PubKey][]core.ParSignedData) error {
+ timesCalled++
+
+ return nil
+ })
+
+ duty := core.NewProposerDuty(uint64(42))
+ pubkey := testutil.RandomCorePubKey(t)
+
+ for i := range 2 {
+ full, err := core.NewPartialVersionedSignedProposal(testutil.RandomDenebVersionedSignedProposal(), i+1)
+ require.NoError(t, err)
+
+ if i == 0 {
+ require.NoError(t, db.StoreInternal(ctx, duty, core.ParSignedDataSet{
+ pubkey: full,
+ }))
+ } else {
+ require.NoError(t, db.StoreExternal(ctx, duty, core.ParSignedDataSet{
+ pubkey: full,
+ }))
+ }
+ }
+
+ require.Zero(t, timesCalled)
+ })
+}
+
 func TestGetThresholdMatching(t *testing.T) {
  const n = 4
 
@@ -149,6 +226,66 @@ func TestMemDBThreshold(t *testing.T) {
  require.Equal(t, 2, timesCalled)
 }
 
+func Test_rootFromParSigDataSet(t *testing.T) {
+ blockData := testutil.RandomDenebVersionedSignedProposal()
+ full, err := core.NewPartialVersionedSignedProposal(blockData, 1)
+ require.NoError(t, err)
+
+ tests := []struct {
+ name string
+ duty core.Duty
+ data core.ParSignedData
+ shouldBeZero bool
+ wantErr bool
+ }{
+ {
+ "ok",
+ core.NewProposerDuty(uint64(42)),
+ full,
+ false,
+ false,
+ },
+ {
+ "sync message",
+ core.NewSyncMessageDuty(uint64(42)),
+ full,
+ true,
+ false,
+ },
+ {
+ "sync contribution",
+ core.NewSyncContributionDuty(uint64(42)),
+ full,
+ true,
+ false,
+ },
+ {
+ "signature",
+ core.NewSignatureDuty(uint64(42)),
+ full,
+ true,
+ false,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := rootFromParSigDataSet(tt.duty, tt.data)
+
+ if tt.shouldBeZero {
+ require.Zero(t, got)
+ } else {
+ require.NotZero(t, got)
+ }
+
+ if tt.wantErr {
+ require.Error(t, err)
+ } else {
+ require.NoError(t, err)
+ }
+ })
+ }
+}
+
 func newTestDeadliner() *testDeadliner {
  return &testDeadliner{
  ch: make(chan core.Duty),

core/tracker/tracker.go

@@ -4,6 +4,7 @@ package tracker
 
 import (
  "context"
+ "encoding/hex"
  "encoding/json"
  "fmt"
 
@@ -104,10 +105,19 @@ type Tracker struct {
 
  // participationReporter instruments duty peer participation.
  participationReporter func(ctx context.Context, duty core.Duty, failed bool, participatedShares map[int]int, unexpectedPeers map[int]int, expectedPerPeer int)
+
+ // peers is the set of peers contained in the cluster.
+ peers map[int]p2p.Peer
 }
 
 // New returns a new Tracker. The deleter deadliner must return well after analyser deadliner since duties of the same slot are often analysed together.
 func New(analyser core.Deadliner, deleter core.Deadliner, peers []p2p.Peer, fromSlot uint64) *Tracker {
+ pm := make(map[int]p2p.Peer, len(peers))
+
+ for _, p := range peers {
+ pm[p.Index] = p
+ }
+
  t := &Tracker{
  input: make(chan event),
  events: make(map[core.Duty][]event),
@@ -118,6 +128,7 @@ func New(analyser core.Deadliner, deleter core.Deadliner, peers []p2p.Peer, from
  parSigReporter: reportParSigs,
  failedDutyReporter: newFailedDutyReporter(),
  participationReporter: newParticipationReporter(peers),
+ peers: pm,
  }
 
  return t
@@ -156,6 +167,26 @@ func (t *Tracker) Run(ctx context.Context) error {
  continue // Ignore unsupported duties
  }
 
+ // For parsigdb inconsistent reason, print what roots where generated by what peer
+ if failed && reason == reasonBugParSigDBInconsistent {
+ for pubkey, parsig := range parsigs {
+ for root, partials := range parsig {
+ var peers []string
+ for _, p := range partials {
+ peers = append(peers, t.peers[p.ShareIdx].Name)
+ }
+
+ log.Debug(
+ ctx,
+ "Detected inconsistent ParSigDB data for duty",
+ z.Str("pubkey", pubkey.String()),
+ z.Str("root", hex.EncodeToString(root[:])),
+ z.Any("peers", peers),
+ )
+ }
+ }
+ }
+
  t.failedDutyReporter(ctx, duty, failed, failedStep, reason, failedErr)
 
  // Analyse peer participation