Skip to content

Commit e90a07b

Browse files
gupadhyayaGanesha Upadhyaya
gupadhyaya
and
Ganesha Upadhyaya
authored
fraud proof processing using go-fraud package (#928)
Update existing fraud proof processing logic to use ProofService from go-fraud package. Also, removes the old fraud proof gossip logic. Fixes #895 & #945 --------- Co-authored-by: Ganesha Upadhyaya <gupadhyaya@Ganeshas-MacBook-Pro-2.local>
1 parent c769503 commit e90a07b

8 files changed

Lines changed: 231 additions & 139 deletions

File tree

‎block/manager.go‎

Lines changed: 57 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"sync/atomic"
1010
"time"
1111

12+
"github.com/celestiaorg/go-fraud/fraudserv"
1213
"github.com/libp2p/go-libp2p/core/crypto"
1314
abci "github.com/tendermint/tendermint/abci/types"
1415
tmcrypto "github.com/tendermint/tendermint/crypto"
@@ -60,8 +61,6 @@ type Manager struct {
6061

6162
HeaderCh chan *types.SignedHeader
6263

63-
FraudProofInCh chan *abci.FraudProof
64-
6564
blockInCh chan newBlockEvent
6665
syncCache map[uint64]*types.Block
6766

@@ -153,7 +152,6 @@ func NewManager(
153152
// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
154153
HeaderCh: make(chan *types.SignedHeader, 100),
155154
blockInCh: make(chan newBlockEvent, 100),
156-
FraudProofInCh: make(chan *abci.FraudProof, 100),
157155
retrieveMtx: new(sync.Mutex),
158156
lastStateMtx: new(sync.Mutex),
159157
syncCache: make(map[uint64]*types.Block),
@@ -181,10 +179,6 @@ func (m *Manager) SetDALC(dalc da.DataAvailabilityLayerClient) {
181179
m.retriever = dalc.(da.BlockRetriever)
182180
}
183181

184-
func (m *Manager) GetFraudProofOutChan() chan *abci.FraudProof {
185-
return m.executor.FraudProofOutCh
186-
}
187-
188182
// AggregationLoop is responsible for aggregating transactions into rollup-blocks.
189183
func (m *Manager) AggregationLoop(ctx context.Context, lazy bool) {
190184
initialHeight := uint64(m.genesis.InitialHeight)
@@ -249,6 +243,57 @@ func (m *Manager) AggregationLoop(ctx context.Context, lazy bool) {
249243
}
250244
}
251245

246+
func (m *Manager) SetFraudProofService(fraudProofServ *fraudserv.ProofService) {
247+
m.executor.SetFraudProofService(fraudProofServ)
248+
}
249+
250+
func (m *Manager) ProcessFraudProof(ctx context.Context, cancel context.CancelFunc) {
251+
// subscribe to state fraud proof
252+
sub, err := m.executor.FraudService.Subscribe(types.StateFraudProofType)
253+
if err != nil {
254+
m.logger.Error("failed to subscribe to fraud proof gossip", "error", err)
255+
return
256+
}
257+
defer sub.Cancel()
258+
259+
// continuously process the fraud proofs received via subscription
260+
for {
261+
// sub.Proof is a blocking call that only returns on proof received or context ended
262+
proof, err := sub.Proof(ctx)
263+
if err != nil {
264+
m.logger.Error("failed to receive gossiped fraud proof", "error", err)
265+
return
266+
}
267+
268+
// only handle the state fraud proofs for now
269+
fraudProof, ok := proof.(*types.StateFraudProof)
270+
if !ok {
271+
m.logger.Error("unexpected type received for state fraud proof", "error", err)
272+
return
273+
}
274+
m.logger.Debug("fraud proof received",
275+
"block height", fraudProof.BlockHeight,
276+
"pre-state app hash", fraudProof.PreStateAppHash,
277+
"expected valid app hash", fraudProof.ExpectedValidAppHash,
278+
"length of state witness", len(fraudProof.StateWitness),
279+
)
280+
// TODO(light-client): Set up a new cosmos-sdk app
281+
// TODO: Add fraud proof window validation
282+
283+
success, err := m.executor.VerifyFraudProof(&fraudProof.FraudProof, fraudProof.ExpectedValidAppHash)
284+
if err != nil {
285+
m.logger.Error("failed to verify fraud proof", "error", err)
286+
continue
287+
}
288+
if success {
289+
// halt chain
290+
m.logger.Info("verified fraud proof, halting chain")
291+
cancel()
292+
return
293+
}
294+
}
295+
}
296+
252297
// SyncLoop is responsible for syncing blocks.
253298
//
254299
// 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) {
277322
if err != nil {
278323
m.logger.Info("failed to sync next block", "error", err)
279324
}
280-
case fraudProof := <-m.FraudProofInCh:
281-
m.logger.Debug("fraud proof received",
282-
"block height", fraudProof.BlockHeight,
283-
"pre-state app hash", fraudProof.PreStateAppHash,
284-
"expected valid app hash", fraudProof.ExpectedValidAppHash,
285-
"length of state witness", len(fraudProof.StateWitness),
286-
)
287-
// TODO(light-client): Set up a new cosmos-sdk app
288-
// TODO: Add fraud proof window validation
289-
290-
success, err := m.executor.VerifyFraudProof(fraudProof, fraudProof.ExpectedValidAppHash)
291-
if err != nil {
292-
m.logger.Error("failed to verify fraud proof", "error", err)
293-
continue
294-
}
295-
if success {
296-
// halt chain
297-
m.logger.Info("verified fraud proof, halting chain")
298-
cancel()
299-
return
300-
}
301-
302325
case <-ctx.Done():
303326
return
304327
}
@@ -338,7 +361,6 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
338361
if err != nil {
339362
return fmt.Errorf("failed to Commit: %w", err)
340363
}
341-
m.store.SetHeight(uint64(b.SignedHeader.Header.Height()))
342364

343365
err = m.store.SaveBlockResponses(uint64(b.SignedHeader.Header.Height()), responses)
344366
if err != nil {
@@ -351,6 +373,8 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
351373
return err
352374
}
353375

376+
m.store.SetHeight(uint64(b.SignedHeader.Header.Height()))
377+
354378
if daHeight > newState.DAHeight {
355379
newState.DAHeight = daHeight
356380
}
@@ -568,9 +592,6 @@ func (m *Manager) publishBlock(ctx context.Context) error {
568592

569593
blockHeight := uint64(block.SignedHeader.Header.Height())
570594

571-
// Only update the stored height after successfully submitting to DA layer and committing to the DB
572-
m.store.SetHeight(blockHeight)
573-
574595
// Commit the new state and block which writes to disk on the proxy app
575596
_, _, err = m.executor.Commit(ctx, newState, block, responses)
576597
if err != nil {
@@ -589,6 +610,9 @@ func (m *Manager) publishBlock(ctx context.Context) error {
589610
return err
590611
}
591612

613+
// Only update the stored height after successfully submitting to DA layer and committing to the DB
614+
m.store.SetHeight(blockHeight)
615+
592616
newState.DAHeight = atomic.LoadUint64(&m.daHeight)
593617
// After this call m.lastState is the NEW state returned from ApplyBlock
594618
m.lastState = newState

‎node/full.go‎

Lines changed: 4 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@ import (
3434
"github.com/rollkit/rollkit/state/txindex"
3535
"github.com/rollkit/rollkit/state/txindex/kv"
3636
"github.com/rollkit/rollkit/store"
37-
"github.com/rollkit/rollkit/types"
3837
)
3938

4039
// prefixes used in KV store to separate main node data from DALC data
@@ -173,7 +172,7 @@ func newFullNode(
173172
},
174173
mainKV,
175174
true,
176-
types.StateFraudProofType,
175+
genesis.ChainID,
177176
)
178177

179178
ctx, cancel := context.WithCancel(ctx)
@@ -203,7 +202,6 @@ func newFullNode(
203202
node.BaseService = *service.NewBaseService(logger, "Node", node)
204203

205204
node.P2P.SetTxValidator(node.newTxValidator())
206-
node.P2P.SetFraudProofValidator(node.newFraudProofValidator())
207205

208206
return node, nil
209207
}
@@ -248,27 +246,6 @@ func (n *FullNode) headerPublishLoop(ctx context.Context) {
248246
}
249247
}
250248

251-
func (n *FullNode) fraudProofPublishLoop(ctx context.Context) {
252-
for {
253-
select {
254-
case fraudProof := <-n.blockManager.GetFraudProofOutChan():
255-
n.Logger.Info("generated fraud proof: ", fraudProof.String())
256-
fraudProofBytes, err := fraudProof.Marshal()
257-
if err != nil {
258-
panic(fmt.Errorf("failed to serialize fraud proof: %w", err))
259-
}
260-
n.Logger.Info("gossipping fraud proof...")
261-
err = n.P2P.GossipFraudProof(context.Background(), fraudProofBytes)
262-
if err != nil {
263-
n.Logger.Error("failed to gossip fraud proof", "error", err)
264-
}
265-
_ = n.Stop()
266-
case <-ctx.Done():
267-
return
268-
}
269-
}
270-
}
271-
272249
// OnStart is a part of Service interface.
273250
func (n *FullNode) OnStart() error {
274251

@@ -292,16 +269,16 @@ func (n *FullNode) OnStart() error {
292269
if err = n.fraudService.Start(n.ctx); err != nil {
293270
return fmt.Errorf("error while starting fraud exchange service: %w", err)
294271
}
272+
n.blockManager.SetFraudProofService(n.fraudService)
295273

296274
if n.conf.Aggregator {
297275
n.Logger.Info("working in aggregator mode", "block time", n.conf.BlockTime)
298276
go n.blockManager.AggregationLoop(n.ctx, n.conf.LazyAggregator)
299277
go n.headerPublishLoop(n.ctx)
300278
}
279+
go n.blockManager.ProcessFraudProof(n.ctx, n.cancel)
301280
go n.blockManager.RetrieveLoop(n.ctx)
302281
go n.blockManager.SyncLoop(n.ctx, n.cancel)
303-
go n.fraudProofPublishLoop(n.ctx)
304-
305282
return nil
306283
}
307284

@@ -326,6 +303,7 @@ func (n *FullNode) OnStop() {
326303
err := n.dalc.Stop()
327304
err = multierr.Append(err, n.P2P.Close())
328305
err = multierr.Append(err, n.hExService.Stop())
306+
err = multierr.Append(err, n.fraudService.Stop(n.ctx))
329307
n.Logger.Error("errors while stopping node:", "errors", err)
330308
}
331309

@@ -384,23 +362,6 @@ func (n *FullNode) newTxValidator() p2p.GossipValidator {
384362
}
385363
}
386364

387-
// newFraudProofValidator returns a pubsub validator that validates a fraud proof and forwards
388-
// it to be verified
389-
func (n *FullNode) newFraudProofValidator() p2p.GossipValidator {
390-
return func(fraudProofMsg *p2p.GossipMessage) bool {
391-
n.Logger.Debug("fraud proof received", "from", fraudProofMsg.From, "bytes", len(fraudProofMsg.Data))
392-
fraudProof := abci.FraudProof{}
393-
err := fraudProof.Unmarshal(fraudProofMsg.Data)
394-
if err != nil {
395-
n.Logger.Error("failed to deserialize fraud proof", "error", err)
396-
return false
397-
}
398-
// TODO(manav): Add validation checks for fraud proof here
399-
n.blockManager.FraudProofInCh <- &fraudProof
400-
return true
401-
}
402-
}
403-
404365
func newPrefixKV(kvStore ds.Datastore, prefix string) ds.TxnDatastore {
405366
return (ktds.Wrap(kvStore, ktds.PrefixTransform{Prefix: ds.NewKey(prefix)}).Children()[0]).(ds.TxnDatastore)
406367
}

‎node/full_node_integration_test.go‎

Lines changed: 113 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -353,6 +353,118 @@ func testSingleAggreatorSingleFullNodeSingleLightNode(t *testing.T) {
353353
assert.Equal(n1h, n3h, "heights must match")
354354
}
355355

356+
func testSingleAggreatorSingleFullNodeFraudProofGossip(t *testing.T) {
357+
assert := assert.New(t)
358+
require := require.New(t)
359+
360+
var wg sync.WaitGroup
361+
aggCtx, aggCancel := context.WithCancel(context.Background())
362+
ctx, cancel := context.WithCancel(context.Background())
363+
clientNodes := 1
364+
nodes, apps := createNodes(aggCtx, ctx, clientNodes+1, true, &wg, t)
365+
366+
for _, app := range apps {
367+
app.On("VerifyFraudProof", mock.Anything).Return(abci.ResponseVerifyFraudProof{Success: true}).Run(func(args mock.Arguments) {
368+
wg.Done()
369+
}).Once()
370+
}
371+
372+
aggNode := nodes[0]
373+
fullNode := nodes[1]
374+
375+
wg.Add(clientNodes + 1)
376+
require.NoError(aggNode.Start())
377+
time.Sleep(2 * time.Second)
378+
require.NoError(fullNode.Start())
379+
380+
wg.Wait()
381+
// aggregator should have 0 GenerateFraudProof calls and 1 VerifyFraudProof calls
382+
apps[0].AssertNumberOfCalls(t, "GenerateFraudProof", 0)
383+
apps[0].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
384+
// fullnode should have 1 GenerateFraudProof calls and 1 VerifyFraudProof calls
385+
apps[1].AssertNumberOfCalls(t, "GenerateFraudProof", 1)
386+
apps[1].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
387+
388+
n1Frauds, err := aggNode.fraudService.Get(aggCtx, types.StateFraudProofType)
389+
require.NoError(err)
390+
aggCancel()
391+
require.NoError(aggNode.Stop())
392+
393+
n2Frauds, err := fullNode.fraudService.Get(aggCtx, types.StateFraudProofType)
394+
require.NoError(err)
395+
cancel()
396+
require.NoError(fullNode.Stop())
397+
398+
assert.Equal(len(n1Frauds), 1, "number of fraud proofs received via gossip should be 1")
399+
assert.Equal(len(n2Frauds), 1, "number of fraud proofs received via gossip should be 1")
400+
assert.Equal(n1Frauds, n2Frauds, "the received fraud proofs after gossip must match")
401+
}
402+
403+
func testSingleAggreatorTwoFullNodeFraudProofSync(t *testing.T) {
404+
assert := assert.New(t)
405+
require := require.New(t)
406+
407+
var wg sync.WaitGroup
408+
aggCtx, aggCancel := context.WithCancel(context.Background())
409+
ctx, cancel := context.WithCancel(context.Background())
410+
clientNodes := 2
411+
nodes, apps := createNodes(aggCtx, ctx, clientNodes+1, true, &wg, t)
412+
413+
for _, app := range apps {
414+
app.On("VerifyFraudProof", mock.Anything).Return(abci.ResponseVerifyFraudProof{Success: true}).Run(func(args mock.Arguments) {
415+
wg.Done()
416+
}).Once()
417+
}
418+
419+
aggNode := nodes[0]
420+
fullNode1 := nodes[1]
421+
fullNode2 := nodes[2]
422+
423+
wg.Add(clientNodes)
424+
require.NoError(aggNode.Start())
425+
time.Sleep(2 * time.Second)
426+
require.NoError(fullNode1.Start())
427+
428+
wg.Wait()
429+
// aggregator should have 0 GenerateFraudProof calls and 1 VerifyFraudProof calls
430+
apps[0].AssertNumberOfCalls(t, "GenerateFraudProof", 0)
431+
apps[0].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
432+
// fullnode1 should have 1 GenerateFraudProof calls and 1 VerifyFraudProof calls
433+
apps[1].AssertNumberOfCalls(t, "GenerateFraudProof", 1)
434+
apps[1].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
435+
436+
n1Frauds, err := aggNode.fraudService.Get(aggCtx, types.StateFraudProofType)
437+
require.NoError(err)
438+
439+
n2Frauds, err := fullNode1.fraudService.Get(aggCtx, types.StateFraudProofType)
440+
require.NoError(err)
441+
assert.Equal(n1Frauds, n2Frauds, "number of fraud proofs gossiped between nodes must match")
442+
443+
wg.Add(1)
444+
// delay start node3 such that it can sync the fraud proof from peers, instead of listening to gossip
445+
require.NoError(fullNode2.Start())
446+
447+
wg.Wait()
448+
// fullnode2 should have 1 GenerateFraudProof calls and 1 VerifyFraudProof calls
449+
apps[2].AssertNumberOfCalls(t, "GenerateFraudProof", 1)
450+
apps[2].AssertNumberOfCalls(t, "VerifyFraudProof", 1)
451+
452+
n3Frauds, err := fullNode2.fraudService.Get(aggCtx, types.StateFraudProofType)
453+
require.NoError(err)
454+
assert.Equal(n1Frauds, n3Frauds, "number of fraud proofs gossiped between nodes must match")
455+
456+
aggCancel()
457+
require.NoError(aggNode.Stop())
458+
cancel()
459+
require.NoError(fullNode1.Stop())
460+
require.NoError(fullNode2.Stop())
461+
}
462+
463+
func TestFraudProofService(t *testing.T) {
464+
testSingleAggreatorSingleFullNodeFraudProofGossip(t)
465+
testSingleAggreatorTwoFullNodeFraudProofSync(t)
466+
}
467+
356468
// TODO: rewrite this integration test to accommodate gossip/halting mechanism of full nodes after fraud proof generation (#693)
357469
// TestFraudProofTrigger setups a network of nodes, with single malicious aggregator and multiple producers.
358470
// 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
542654
}
543655

544656
if isMalicious && !aggregator {
545-
app.On("GenerateFraudProof", mock.Anything).Return(abci.ResponseGenerateFraudProof{FraudProof: &abci.FraudProof{}})
657+
app.On("GenerateFraudProof", mock.Anything).Return(abci.ResponseGenerateFraudProof{FraudProof: &abci.FraudProof{BlockHeight: 1, FraudulentBeginBlock: &abci.RequestBeginBlock{Hash: []byte("123")}, ExpectedValidAppHash: nonMaliciousAppHash}})
546658
}
547659
app.On("DeliverTx", mock.Anything).Return(abci.ResponseDeliverTx{}).Run(func(args mock.Arguments) {
548660
wg.Done()

0 commit comments

Comments
 (0)