Fix indexed attestations Slasher store (#3897)

* first version of the watchtower api

* service files

* Begin work on grpc server

* More changes to server

* REnames and mock setup

* working test

* merge

* double propose detection test

* nishant review

* todo change

* gaz

* fix service

* gaz

* remove unused import

* gaz

* resolve circular dependency

* resolve circular dependency 2nd try

* remove package

* fix package

* fix test

* added tests

* gaz

* remove status check

* gaz

* remove context

* remove context

* change var name

* moved to rpc dir

* gaz

* remove server code

* gaz

* slasher server

* visibility change

* pb

* service update

* gaz

* slasher grpc server

* making it work

* setup db and start

* gaz

* service flags fixes

* grpc service running

* go imports

* remove new initializer

* gaz

* remove feature flags

* change back SetupSlasherDB

* fix SetupSlasherDB calls

* define err

* fix bad merge

* fix test

* fix imports

* fix imports

* fix imports

* add cancel

* comment stop

* fix cancel issue

* remove unneeded code

* bring back bad merge that removed TODO

* remove use of epoch as am input

* fixed slasher to be runable again

* wait for channel close

* gaz

* small test

* flags fix

* fix flag order

slasher/db/indexed_attestations.go

@@ -34,9 +34,9 @@ func createValidatorIDsToIndexedAttestationList(enc []byte) (*ethpb.ValidatorIDT
 // IndexedAttestation accepts a epoch and validator index and returns a list of
 // indexed attestations.
 // Returns nil if the indexed attestation does not exist.
-func (db *Store) IndexedAttestation(epoch uint64, validatorID uint64) ([]*ethpb.IndexedAttestation, error) {
+func (db *Store) IndexedAttestation(sourceEpoch uint64, targetEpoch uint64, validatorID uint64) ([]*ethpb.IndexedAttestation, error) {
 	var iAtt []*ethpb.IndexedAttestation
-	key := bytesutil.Bytes8(epoch)
+	key := append(bytesutil.Bytes8(sourceEpoch), bytesutil.Bytes8(targetEpoch)...)
 	err := db.view(func(tx *bolt.Tx) error {
 		bucket := tx.Bucket(indexedAttestationsIndicesBucket)
 		enc := bucket.Get(key)
@@ -48,7 +48,7 @@ func (db *Store) IndexedAttestation(epoch uint64, validatorID uint64) ([]*ethpb.
 			i := sort.Search(len(a.Indices), func(i int) bool { return a.Indices[i] >= validatorID })
 			if i < len(a.Indices) && a.Indices[i] == validatorID {
 				iaBucket := tx.Bucket(historicIndexedAttestationsBucket)
-				key := encodeEpochSig(epoch, a.Signature)
+				key := encodeEpochSig(sourceEpoch, targetEpoch, a.Signature)
 				enc = iaBucket.Get(key)
 				if len(enc) == 0 {
 					continue
@@ -67,8 +67,8 @@ func (db *Store) IndexedAttestation(epoch uint64, validatorID uint64) ([]*ethpb.
 }
 
 // HasIndexedAttestation accepts an epoch and validator id and returns true if the indexed attestation exists.
-func (db *Store) HasIndexedAttestation(epoch uint64, validatorID uint64) bool {
-	key := bytesutil.Bytes8(epoch)
+func (db *Store) HasIndexedAttestation(sourceEpoch uint64, targetEpoch uint64, validatorID uint64) bool {
+	key := append(bytesutil.Bytes8(sourceEpoch), bytesutil.Bytes8(targetEpoch)...)
 	var hasAttestation bool
 	// #nosec G104
 	_ = db.view(func(tx *bolt.Tx) error {
@@ -93,8 +93,8 @@ func (db *Store) HasIndexedAttestation(epoch uint64, validatorID uint64) bool {
 }
 
 // SaveIndexedAttestation accepts epoch and indexed attestation and writes it to disk.
-func (db *Store) SaveIndexedAttestation(epoch uint64, idxAttestation *ethpb.IndexedAttestation) error {
-	key := encodeEpochSig(epoch, idxAttestation.Signature)
+func (db *Store) SaveIndexedAttestation(idxAttestation *ethpb.IndexedAttestation) error {
+	key := encodeEpochSig(idxAttestation.Data.Source.Epoch, idxAttestation.Data.Target.Epoch, idxAttestation.Signature)
 	enc, err := proto.Marshal(idxAttestation)
 	if err != nil {
 		return errors.Wrap(err, "failed to marshal")
@@ -107,7 +107,7 @@ func (db *Store) SaveIndexedAttestation(epoch uint64, idxAttestation *ethpb.Inde
 		if val != nil {
 			return nil
 		}
-		createIndexedAttestationIndicesFromData(epoch, idxAttestation, tx)
+		createIndexedAttestationIndicesFromData(idxAttestation, tx)
 		if err := bucket.Put(key, enc); err != nil {
 			return errors.Wrap(err, "failed to include the indexed attestation in the historic indexed attestation bucket")
 		}
@@ -116,25 +116,27 @@ func (db *Store) SaveIndexedAttestation(epoch uint64, idxAttestation *ethpb.Inde
 	})
 
 	// prune history to max size every 10th epoch
-	if epoch%10 == 0 {
+	if idxAttestation.Data.Source.Epoch%10 == 0 {
 		weakSubjectivityPeriod := uint64(54000)
-		err = db.PruneHistory(epoch, weakSubjectivityPeriod)
+		err = db.PruneHistory(idxAttestation.Data.Source.Epoch, weakSubjectivityPeriod)
 	}
 	return err
 }
 
-func createIndexedAttestationIndicesFromData(epoch uint64, idxAttestation *ethpb.IndexedAttestation, tx *bolt.Tx) error {
+func createIndexedAttestationIndicesFromData(idxAttestation *ethpb.IndexedAttestation, tx *bolt.Tx) error {
 	indices := append(idxAttestation.CustodyBit_0Indices, idxAttestation.CustodyBit_1Indices...)
-	dataRoot, err := ssz.Marshal(idxAttestation.Data)
+	dataRoot, err := ssz.HashTreeRoot(idxAttestation.Data)
+
 	if err != nil {
-		return errors.Wrap(err, "failed to marshal indexed attestation data.")
+		return errors.Wrap(err, "failed to hash indexed attestation data.")
 	}
 	protoIdxAtt := &ethpb.ValidatorIDToIdxAtt{
 		Signature: idxAttestation.Signature,
 		Indices:   indices,
-		DataRoot:  dataRoot,
+		DataRoot:  dataRoot[:],
 	}
-	key := bytesutil.Bytes8(epoch)
+	key := append(bytesutil.Bytes8(idxAttestation.Data.Source.Epoch), bytesutil.Bytes8(idxAttestation.Data.Target.Epoch)...)
+
 	bucket := tx.Bucket(indexedAttestationsIndicesBucket)
 	enc := bucket.Get(key)
 	vIdxList, err := createValidatorIDsToIndexedAttestationList(enc)
@@ -153,31 +155,32 @@ func createIndexedAttestationIndicesFromData(epoch uint64, idxAttestation *ethpb
 }
 
 // DeleteIndexedAttestation deletes a indexed attestation using the slot and its root as keys in their respective buckets.
-func (db *Store) DeleteIndexedAttestation(epoch uint64, idxAttestation *ethpb.IndexedAttestation) error {
-	key := encodeEpochSig(epoch, idxAttestation.Signature)
+func (db *Store) DeleteIndexedAttestation(idxAttestation *ethpb.IndexedAttestation) error {
+	key := encodeEpochSig(idxAttestation.Data.Source.Epoch, idxAttestation.Data.Target.Epoch, idxAttestation.Signature)
 	return db.update(func(tx *bolt.Tx) error {
 		bucket := tx.Bucket(historicIndexedAttestationsBucket)
 		enc := bucket.Get(key)
 		if enc == nil {
 			return nil
 		}
-		removeIndexedAttestationIndicesFromData(epoch, idxAttestation, tx)
+		removeIndexedAttestationIndicesFromData(idxAttestation, tx)
 		if err := bucket.Delete(key); err != nil {
+			tx.Rollback()
 			return errors.Wrap(err, "failed to delete the indexed attestation from historic indexed attestation bucket")
 		}
 		return nil
 	})
 }
 
-func removeIndexedAttestationIndicesFromData(epoch uint64, idxAttestation *ethpb.IndexedAttestation, tx *bolt.Tx) error {
+func removeIndexedAttestationIndicesFromData(idxAttestation *ethpb.IndexedAttestation, tx *bolt.Tx) error {
 	indices := append(idxAttestation.CustodyBit_0Indices, idxAttestation.CustodyBit_1Indices...)
-	dataRoot, err := ssz.Marshal(idxAttestation.Data)
+	dataRoot, err := ssz.HashTreeRoot(idxAttestation.Data)
 	protoIdxAtt := &ethpb.ValidatorIDToIdxAtt{
 		Signature: idxAttestation.Signature,
 		Indices:   indices,
-		DataRoot:  dataRoot,
+		DataRoot:  dataRoot[:],
 	}
-	key := bytesutil.Bytes8(epoch)
+	key := append(bytesutil.Bytes8(idxAttestation.Data.Source.Epoch), bytesutil.Bytes8(idxAttestation.Data.Target.Epoch)...)
 	bucket := tx.Bucket(indexedAttestationsIndicesBucket)
 	enc := bucket.Get(key)
 	vIdxList, err := createValidatorIDsToIndexedAttestationList(enc)

slasher/db/indexed_attestations_test.go

@@ -8,8 +8,7 @@ import (
 )
 
 type testStruct struct {
-	epoch uint64
-	iA    *ethpb.IndexedAttestation
+	iA *ethpb.IndexedAttestation
 }
 
 var tests []testStruct
@@ -17,30 +16,27 @@ var tests []testStruct
 func init() {
 	tests = []testStruct{
 		{
-			epoch: uint64(0),
 			iA: &ethpb.IndexedAttestation{Signature: []byte("let me in"), CustodyBit_0Indices: []uint64{0}, Data: &ethpb.AttestationData{
 				Source: &ethpb.Checkpoint{Epoch: 0},
-				Target: &ethpb.Checkpoint{Epoch: 0},
+				Target: &ethpb.Checkpoint{Epoch: 1},
 				Crosslink: &ethpb.Crosslink{
 					Shard: 4,
 				},
 			}},
 		},
 		{
-			epoch: uint64(0),
 			iA: &ethpb.IndexedAttestation{Signature: []byte("let me in 2nd"), CustodyBit_0Indices: []uint64{1, 2}, Data: &ethpb.AttestationData{
 				Source: &ethpb.Checkpoint{Epoch: 0},
-				Target: &ethpb.Checkpoint{Epoch: 0},
+				Target: &ethpb.Checkpoint{Epoch: 2},
 				Crosslink: &ethpb.Crosslink{
 					Shard: 4,
 				},
 			}},
 		},
 		{
-			epoch: uint64(1),
 			iA: &ethpb.IndexedAttestation{Signature: []byte("let me in 3rd"), CustodyBit_0Indices: []uint64{0}, Data: &ethpb.AttestationData{
-				Source: &ethpb.Checkpoint{Epoch: 0},
-				Target: &ethpb.Checkpoint{Epoch: 0},
+				Source: &ethpb.Checkpoint{Epoch: 1},
+				Target: &ethpb.Checkpoint{Epoch: 2},
 				Crosslink: &ethpb.Crosslink{
 					Shard: 4,
 				},
@@ -56,12 +52,12 @@ func TestNilDBHistoryIdxAtt(t *testing.T) {
 	epoch := uint64(1)
 	validatorID := uint64(1)
 
-	hasIdxAtt := db.HasIndexedAttestation(epoch, validatorID)
+	hasIdxAtt := db.HasIndexedAttestation(epoch, epoch, validatorID)
 	if hasIdxAtt {
 		t.Fatal("HasIndexedAttestation should return false")
 	}
 
-	idxAtt, err := db.IndexedAttestation(epoch, validatorID)
+	idxAtt, err := db.IndexedAttestation(epoch, epoch, validatorID)
 	if err != nil {
 		t.Fatalf("failed to get indexed attestation: %v", err)
 	}
@@ -75,12 +71,12 @@ func TestSaveIdxAtt(t *testing.T) {
 	defer TeardownSlasherDB(t, db)
 
 	for _, tt := range tests {
-		err := db.SaveIndexedAttestation(tt.epoch, tt.iA)
+		err := db.SaveIndexedAttestation(tt.iA)
 		if err != nil {
 			t.Fatalf("save indexed attestation failed: %v", err)
 		}
 
-		iAarray, err := db.IndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		iAarray, err := db.IndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 		if err != nil {
 			t.Fatalf("failed to get indexed attestation: %v", err)
 		}
@@ -98,14 +94,14 @@ func TestDeleteHistoryIdxAtt(t *testing.T) {
 
 	for _, tt := range tests {
 
-		err := db.SaveIndexedAttestation(tt.epoch, tt.iA)
+		err := db.SaveIndexedAttestation(tt.iA)
 		if err != nil {
 			t.Fatalf("save indexed attestation failed: %v", err)
 		}
 	}
 
 	for _, tt := range tests {
-		iAarray, err := db.IndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		iAarray, err := db.IndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 		if err != nil {
 			t.Fatalf("failed to get index attestation: %v", err)
 		}
@@ -113,12 +109,12 @@ func TestDeleteHistoryIdxAtt(t *testing.T) {
 		if iAarray == nil || !reflect.DeepEqual(iAarray[0], tt.iA) {
 			t.Fatalf("get should return indexed attestation: %v", iAarray)
 		}
-		err = db.DeleteIndexedAttestation(tt.epoch, tt.iA)
+		err = db.DeleteIndexedAttestation(tt.iA)
 		if err != nil {
 			t.Fatalf("delete index attestation failed: %v", err)
 		}
-		iAarray, err = db.IndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
-		hasA := db.HasIndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		iAarray, err = db.IndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
+		hasA := db.HasIndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -139,18 +135,18 @@ func TestHasIdxAtt(t *testing.T) {
 
 	for _, tt := range tests {
 
-		found := db.HasIndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		found := db.HasIndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 		if found {
 			t.Fatal("has indexed attestation should return false for indexed attestations that are not in db")
 		}
-		err := db.SaveIndexedAttestation(tt.epoch, tt.iA)
+		err := db.SaveIndexedAttestation(tt.iA)
 		if err != nil {
 			t.Fatalf("save indexed attestation failed: %v", err)
 		}
 	}
 	for _, tt := range tests {
 
-		found := db.HasIndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		found := db.HasIndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 
 		if !found {
 			t.Fatal("has indexed attestation should return true")
@@ -163,12 +159,12 @@ func TestPruneHistoryIdxAtt(t *testing.T) {
 	defer TeardownSlasherDB(t, db)
 
 	for _, tt := range tests {
-		err := db.SaveIndexedAttestation(tt.epoch, tt.iA)
+		err := db.SaveIndexedAttestation(tt.iA)
 		if err != nil {
 			t.Fatalf("save indexed attestation failed: %v", err)
 		}
 
-		iAarray, err := db.IndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		iAarray, err := db.IndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 		if err != nil {
 			t.Fatalf("failed to get indexed attestation: %v", err)
 		}
@@ -185,13 +181,13 @@ func TestPruneHistoryIdxAtt(t *testing.T) {
 	}
 
 	for _, tt := range tests {
-		iAarray, err := db.IndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		iAarray, err := db.IndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 		if err != nil {
 			t.Fatalf("failed to get indexed attestation: %v", err)
 		}
-		hasIa := db.HasIndexedAttestation(tt.epoch, tt.iA.CustodyBit_0Indices[0])
+		hasIa := db.HasIndexedAttestation(tt.iA.Data.Source.Epoch, tt.iA.Data.Target.Epoch, tt.iA.CustodyBit_0Indices[0])
 
-		if tt.epoch > currentEpoch-historyToKeep {
+		if tt.iA.Data.Source.Epoch > currentEpoch-historyToKeep {
 			if iAarray == nil || !reflect.DeepEqual(iAarray[0], tt.iA) {
 				t.Fatalf("get should return indexed attestation: %v", iAarray)
 			}

slasher/db/schema.go

@@ -19,6 +19,7 @@ func encodeEpochValidatorIDSig(epoch uint64, validatorID uint64, sig []byte) []b
 	return append(append(bytesutil.Bytes8(epoch), bytesutil.Bytes8(validatorID)...), sig...)
 }
 
-func encodeEpochSig(epoch uint64, sig []byte) []byte {
-	return append(bytesutil.Bytes8(epoch), sig...)
+func encodeEpochSig(sourceEpoch uint64, targetEpoch uint64, sig []byte) []byte {
+	st := append(bytesutil.Bytes8(sourceEpoch), bytesutil.Bytes8(targetEpoch)...)
+	return append(st, sig...)
 }

tools/extractor/main.go

@@ -16,7 +16,6 @@ var (
 	datadir = flag.String("datadir", "", "Path to data directory.")
 
 	state = flag.Uint("state", 0, "Extract state at this slot.")
-
 )
 
 func init() {