fraud proof processing using go-fraud package

block/manager.go

@@ -9,6 +9,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/celestiaorg/go-fraud/fraudserv"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	abci "github.com/tendermint/tendermint/abci/types"
 	tmcrypto "github.com/tendermint/tendermint/crypto"
@@ -60,8 +61,6 @@ type Manager struct {
 
 	HeaderCh chan *types.SignedHeader
 
-	FraudProofInCh chan *abci.FraudProof
-
 	blockInCh chan newBlockEvent
 	syncCache map[uint64]*types.Block
 
@@ -153,7 +152,6 @@ func NewManager(
  // channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
  HeaderCh: make(chan *types.SignedHeader, 100),
  blockInCh: make(chan newBlockEvent, 100),
- FraudProofInCh: make(chan *abci.FraudProof, 100),
  retrieveMtx: new(sync.Mutex),
  lastStateMtx: new(sync.Mutex),
  syncCache: make(map[uint64]*types.Block),
@@ -181,10 +179,6 @@ func (m *Manager) SetDALC(dalc da.DataAvailabilityLayerClient) {
 	m.retriever = dalc.(da.BlockRetriever)
 }
 
-func (m *Manager) GetFraudProofOutChan() chan *abci.FraudProof {
-	return m.executor.FraudProofOutCh
-}
-
 // AggregationLoop is responsible for aggregating transactions into rollup-blocks.
 func (m *Manager) AggregationLoop(ctx context.Context, lazy bool) {
 	initialHeight := uint64(m.genesis.InitialHeight)
@@ -249,6 +243,57 @@ func (m *Manager) AggregationLoop(ctx context.Context, lazy bool) {
 	}
 }
 
+func (m *Manager) SetFraudProofService(fraudProofServ *fraudserv.ProofService) {
+	m.executor.SetFraudProofService(fraudProofServ)
+}
+
+func (m *Manager) ProcessFraudProof(ctx context.Context, cancel context.CancelFunc) {
+	// subscribe to state fraud proof
+	sub, err := m.executor.FraudService.Subscribe(types.StateFraudProofType)
+	if err != nil {
+ m.logger.Error("failed to subscribe to fraud proof gossip", "error", err)
+ return
+	}
+	defer sub.Cancel()
+
+	// continuously process the fraud proofs received via subscription
+	for {
+ // sub.Proof is a blocking call that only returns on proof received or context ended
+ proof, err := sub.Proof(ctx)
+ if err != nil {
+ m.logger.Error("failed to receive gossiped fraud proof", "error", err)
+ return
+ }
+
+ // only handle the state fraud proofs for now
+ fraudProof, ok := proof.(*types.StateFraudProof)
+ if !ok {
+ m.logger.Error("unexpected type received for state fraud proof", "error", err)
+ return
+ }
+ m.logger.Debug("fraud proof received",
+ "block height", fraudProof.BlockHeight,
+ "pre-state app hash", fraudProof.PreStateAppHash,
+ "expected valid app hash", fraudProof.ExpectedValidAppHash,
+ "length of state witness", len(fraudProof.StateWitness),
+ )
+ // TODO(light-client): Set up a new cosmos-sdk app
+ // TODO: Add fraud proof window validation
+
+ success, err := m.executor.VerifyFraudProof(&fraudProof.FraudProof, fraudProof.ExpectedValidAppHash)
+ if err != nil {
+ m.logger.Error("failed to verify fraud proof", "error", err)
+ continue
+ }
+ if success {
+ // halt chain
+ m.logger.Info("verified fraud proof, halting chain")
+ cancel()
+ return
+ }
+	}
+}
+
 // SyncLoop is responsible for syncing blocks.
 //
 // SyncLoop processes headers gossiped in P2p network to know what's the latest block height,
@@ -277,28 +322,6 @@ func (m *Manager) SyncLoop(ctx context.Context, cancel context.CancelFunc) {
  if err != nil {
  m.logger.Info("failed to sync next block", "error", err)
  }
- case fraudProof := <-m.FraudProofInCh:
- m.logger.Debug("fraud proof received",
- "block height", fraudProof.BlockHeight,
- "pre-state app hash", fraudProof.PreStateAppHash,
- "expected valid app hash", fraudProof.ExpectedValidAppHash,
- "length of state witness", len(fraudProof.StateWitness),
- )
- // TODO(light-client): Set up a new cosmos-sdk app
- // TODO: Add fraud proof window validation
-
- success, err := m.executor.VerifyFraudProof(fraudProof, fraudProof.ExpectedValidAppHash)
- if err != nil {
- m.logger.Error("failed to verify fraud proof", "error", err)
- continue
- }
- if success {
- // halt chain
- m.logger.Info("verified fraud proof, halting chain")
- cancel()
- return
- }
-
  case <-ctx.Done():
  return
  }
@@ -338,7 +361,6 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
  if err != nil {
  return fmt.Errorf("failed to Commit: %w", err)
  }
- m.store.SetHeight(uint64(b.SignedHeader.Header.Height()))
 
  err = m.store.SaveBlockResponses(uint64(b.SignedHeader.Header.Height()), responses)
  if err != nil {
@@ -351,6 +373,8 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
  return err
  }
 
+ m.store.SetHeight(uint64(b.SignedHeader.Header.Height()))
+
  if daHeight > newState.DAHeight {
  newState.DAHeight = daHeight
  }
@@ -568,9 +592,6 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 
 	blockHeight := uint64(block.SignedHeader.Header.Height())
 
-	// Only update the stored height after successfully submitting to DA layer and committing to the DB
-	m.store.SetHeight(blockHeight)
-
 	// Commit the new state and block which writes to disk on the proxy app
 	_, _, err = m.executor.Commit(ctx, newState, block, responses)
 	if err != nil {
@@ -589,6 +610,9 @@ func (m *Manager) publishBlock(ctx context.Context) error {
  return err
 	}
 
+	// Only update the stored height after successfully submitting to DA layer and committing to the DB
+	m.store.SetHeight(blockHeight)
+
 	newState.DAHeight = atomic.LoadUint64(&m.daHeight)
 	// After this call m.lastState is the NEW state returned from ApplyBlock
 	m.lastState = newState

node/full.go

@@ -34,7 +34,6 @@ import (
 	"github.com/rollkit/rollkit/state/txindex"
 	"github.com/rollkit/rollkit/state/txindex/kv"
 	"github.com/rollkit/rollkit/store"
-	"github.com/rollkit/rollkit/types"
 )
 
 // prefixes used in KV store to separate main node data from DALC data
@@ -173,7 +172,7 @@ func newFullNode(
  },
  mainKV,
  true,
-types.StateFraudProofType,
+genesis.ChainID,
 	)
 
 	ctx, cancel := context.WithCancel(ctx)
@@ -203,7 +202,6 @@ func newFullNode(
 	node.BaseService = *service.NewBaseService(logger, "Node", node)
 
 	node.P2P.SetTxValidator(node.newTxValidator())
-	node.P2P.SetFraudProofValidator(node.newFraudProofValidator())
 
 	return node, nil
 }
@@ -248,27 +246,6 @@ func (n *FullNode) headerPublishLoop(ctx context.Context) {
 	}
 }
 
-func (n *FullNode) fraudProofPublishLoop(ctx context.Context) {
-	for {
- select {
- case fraudProof := <-n.blockManager.GetFraudProofOutChan():
- n.Logger.Info("generated fraud proof: ", fraudProof.String())
- fraudProofBytes, err := fraudProof.Marshal()
- if err != nil {
- panic(fmt.Errorf("failed to serialize fraud proof: %w", err))
- }
- n.Logger.Info("gossipping fraud proof...")
- err = n.P2P.GossipFraudProof(context.Background(), fraudProofBytes)
- if err != nil {
- n.Logger.Error("failed to gossip fraud proof", "error", err)
- }
- _ = n.Stop()
- case <-ctx.Done():
- return
- }
-	}
-}
-
 // OnStart is a part of Service interface.
 func (n *FullNode) OnStart() error {
 
@@ -292,16 +269,16 @@ func (n *FullNode) OnStart() error {
 	if err = n.fraudService.Start(n.ctx); err != nil {
  return fmt.Errorf("error while starting fraud exchange service: %w", err)
 	}
+	n.blockManager.SetFraudProofService(n.fraudService)
 
 	if n.conf.Aggregator {
  n.Logger.Info("working in aggregator mode", "block time", n.conf.BlockTime)
  go n.blockManager.AggregationLoop(n.ctx, n.conf.LazyAggregator)
  go n.headerPublishLoop(n.ctx)
 	}
+	go n.blockManager.ProcessFraudProof(n.ctx, n.cancel)
 	go n.blockManager.RetrieveLoop(n.ctx)
 	go n.blockManager.SyncLoop(n.ctx, n.cancel)
-	go n.fraudProofPublishLoop(n.ctx)
-
 	return nil
 }
 
@@ -326,6 +303,7 @@ func (n *FullNode) OnStop() {
 	err := n.dalc.Stop()
 	err = multierr.Append(err, n.P2P.Close())
 	err = multierr.Append(err, n.hExService.Stop())
+	err = multierr.Append(err, n.fraudService.Stop(n.ctx))
 	n.Logger.Error("errors while stopping node:", "errors", err)
 }
 
@@ -384,23 +362,6 @@ func (n *FullNode) newTxValidator() p2p.GossipValidator {
 	}
 }
 
-// newFraudProofValidator returns a pubsub validator that validates a fraud proof and forwards
-// it to be verified
-func (n *FullNode) newFraudProofValidator() p2p.GossipValidator {
-	return func(fraudProofMsg *p2p.GossipMessage) bool {
- n.Logger.Debug("fraud proof received", "from", fraudProofMsg.From, "bytes", len(fraudProofMsg.Data))
- fraudProof := abci.FraudProof{}
- err := fraudProof.Unmarshal(fraudProofMsg.Data)
- if err != nil {
- n.Logger.Error("failed to deserialize fraud proof", "error", err)
- return false
- }
- // TODO(manav): Add validation checks for fraud proof here
- n.blockManager.FraudProofInCh <- &fraudProof
- return true
-	}
-}
-
 func newPrefixKV(kvStore ds.Datastore, prefix string) ds.TxnDatastore {
 	return (ktds.Wrap(kvStore, ktds.PrefixTransform{Prefix: ds.NewKey(prefix)}).Children()[0]).(ds.TxnDatastore)
 }

node/full_node_integration_test.go

@@ -353,6 +353,118 @@ func testSingleAggreatorSingleFullNodeSingleLightNode(t *testing.T) {
 	assert.Equal(n1h, n3h, "heights must match")
 }
 
+func testSingleAggreatorSingleFullNodeFraudProofGossip(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	var wg sync.WaitGroup
+	aggCtx, aggCancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.Background())
+	clientNodes := 1
+	nodes, apps := createNodes(aggCtx, ctx, clientNodes+1, true, &wg, t)
+
+	for _, app := range apps {
+ app.On("VerifyFraudProof", mock.Anything).Return(abci.ResponseVerifyFraudProof{Success: true}).Run(func(args mock.Arguments) {
+ wg.Done()
+ }).Once()
+	}
+
+	aggNode := nodes[0]
+	fullNode := nodes[1]
+
+	wg.Add(clientNodes + 1)
+	require.NoError(aggNode.Start())
+	time.Sleep(2 * time.Second)
+	require.NoError(fullNode.Start())
+
+	wg.Wait()
+	// aggregator should have 0 GenerateFraudProof calls and 1 VerifyFraudProof calls
+	apps[0].AssertNumberOfCalls(t, "GenerateFraudProof", 0)
+	apps[0].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
+	// fullnode should have 1 GenerateFraudProof calls and 1 VerifyFraudProof calls
+	apps[1].AssertNumberOfCalls(t, "GenerateFraudProof", 1)
+	apps[1].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
+
+	n1Frauds, err := aggNode.fraudService.Get(aggCtx, types.StateFraudProofType)
+	require.NoError(err)
+	aggCancel()
+	require.NoError(aggNode.Stop())
+
+	n2Frauds, err := fullNode.fraudService.Get(aggCtx, types.StateFraudProofType)
+	require.NoError(err)
+	cancel()
+	require.NoError(fullNode.Stop())
+
+	assert.Equal(len(n1Frauds), 1, "number of fraud proofs received via gossip should be 1")
+	assert.Equal(len(n2Frauds), 1, "number of fraud proofs received via gossip should be 1")
+	assert.Equal(n1Frauds, n2Frauds, "the received fraud proofs after gossip must match")
+}
+
+func testSingleAggreatorTwoFullNodeFraudProofSync(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	var wg sync.WaitGroup
+	aggCtx, aggCancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.Background())
+	clientNodes := 2
+	nodes, apps := createNodes(aggCtx, ctx, clientNodes+1, true, &wg, t)
+
+	for _, app := range apps {
+ app.On("VerifyFraudProof", mock.Anything).Return(abci.ResponseVerifyFraudProof{Success: true}).Run(func(args mock.Arguments) {
+ wg.Done()
+ }).Once()
+	}
+
+	aggNode := nodes[0]
+	fullNode1 := nodes[1]
+	fullNode2 := nodes[2]
+
+	wg.Add(clientNodes)
+	require.NoError(aggNode.Start())
+	time.Sleep(2 * time.Second)
+	require.NoError(fullNode1.Start())
+
+	wg.Wait()
+	// aggregator should have 0 GenerateFraudProof calls and 1 VerifyFraudProof calls
+	apps[0].AssertNumberOfCalls(t, "GenerateFraudProof", 0)
+	apps[0].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
+	// fullnode1 should have 1 GenerateFraudProof calls and 1 VerifyFraudProof calls
+	apps[1].AssertNumberOfCalls(t, "GenerateFraudProof", 1)
+	apps[1].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
+
+	n1Frauds, err := aggNode.fraudService.Get(aggCtx, types.StateFraudProofType)
+	require.NoError(err)
+
+	n2Frauds, err := fullNode1.fraudService.Get(aggCtx, types.StateFraudProofType)
+	require.NoError(err)
+	assert.Equal(n1Frauds, n2Frauds, "number of fraud proofs gossiped between nodes must match")
+
+	wg.Add(1)
+	// delay start node3 such that it can sync the fraud proof from peers, instead of listening to gossip
+	require.NoError(fullNode2.Start())
+
+	wg.Wait()
+	// fullnode2 should have 1 GenerateFraudProof calls and 1 VerifyFraudProof calls
+	apps[2].AssertNumberOfCalls(t, "GenerateFraudProof", 1)
+	apps[2].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
+
+	n3Frauds, err := fullNode2.fraudService.Get(aggCtx, types.StateFraudProofType)
+	require.NoError(err)
+	assert.Equal(n1Frauds, n3Frauds, "number of fraud proofs gossiped between nodes must match")
+
+	aggCancel()
+	require.NoError(aggNode.Stop())
+	cancel()
+	require.NoError(fullNode1.Stop())
+	require.NoError(fullNode2.Stop())
+}
+
+func TestFraudProofService(t *testing.T) {
+	testSingleAggreatorSingleFullNodeFraudProofGossip(t)
+	testSingleAggreatorTwoFullNodeFraudProofSync(t)
+}
+
 // TODO: rewrite this integration test to accommodate gossip/halting mechanism of full nodes after fraud proof generation (#693)
 // TestFraudProofTrigger setups a network of nodes, with single malicious aggregator and multiple producers.
 // Aggregator node should produce malicious blocks, nodes should detect fraud, and generate fraud proofs
@@ -542,7 +654,7 @@ func createNode(ctx context.Context, n int, isMalicious bool, aggregator bool, i
 	}
 
 	if isMalicious && !aggregator {
- app.On("GenerateFraudProof", mock.Anything).Return(abci.ResponseGenerateFraudProof{FraudProof: &abci.FraudProof{}})
+ app.On("GenerateFraudProof", mock.Anything).Return(abci.ResponseGenerateFraudProof{FraudProof: &abci.FraudProof{BlockHeight: 1, FraudulentBeginBlock: &abci.RequestBeginBlock{Hash: []byte("123")}, ExpectedValidAppHash: nonMaliciousAppHash}})
 	}
 	app.On("DeliverTx", mock.Anything).Return(abci.ResponseDeliverTx{}).Run(func(args mock.Arguments) {
  wg.Done()