IPC client timeout

README.md

@@ -7,6 +7,11 @@ The FGP is a partition under Unicity BFT.
 It is a headers-only partition (with no units or transactions, and no state or block data other 
 than the hash of the PoW block being finalized).
 
+The finality gadget operates by periodically querying the underlying Proof-of-Work (PoW) `unicity-node` to fetch the
+current best block header. By default, every 2.4 hours, the FGP leader proposes a new state transition that includes the
+hash of this PoW block. Once certified by the BFT partition via a Unicity Certificate (UC), this PoW block is considered
+finalized and added to the partition state, preventing deep reorgs on the base layer.
+
 See Sec 6.3 Finality Gadget of the YP for more details.
 
 ## Prerequisites
@@ -68,3 +73,19 @@ Or via Docker:
 ```bash
 docker run unicity-fgp:local --help
 ```
+
+### Example Run Command
+
+The following is an example of how to start an FGP node locally. By default, the node assumes the underlying PoW socket
+is located at `$HOME/.unicity/node.sock` (configurable via `--pow-socket-path`).
+
+```bash
+build/fgp run \
+    --home "test-nodes/fgp1" \
+    --trust-base test-nodes/trust-base.json \
+    --shard-conf test-nodes/shard-conf-3_0.json \
+    --address "/ip4/127.0.0.1/tcp/30666" \
+    --bootnodes "/ip4/127.0.0.1/tcp/26662/p2p/16Uiu2HAmQMpRWSCskWgsHnAPqHCcUHqe9hHS3Sxta3ziAsz7yU1h" \
+    --log-format text \
+    --log-level info
+```

partition/node.go

@@ -60,8 +60,9 @@ type (
 	}
 
 	roundTimer struct {
-		stop  atomic.Value
-		event chan struct{}
+		isRunning atomic.Bool
+		cancel    context.CancelFunc
+		event     chan struct{}
 	}
 
 	// Node represents a member in the partition and implements an instance of a specific TransactionSystem.
@@ -151,6 +152,9 @@ func (n *Node) loop(ctx context.Context) error {
 		case <-ticker.C:
 			n.handleMonitoring(ctx, lastUCReceived, lastBlockReceived)
 		}
+
+		// central location to manage T1 timeout
+		n.ensureT1TimerState(ctx)
 	}
 }
 

partition/node_init.go

@@ -40,7 +40,6 @@ func NewNode(ctx context.Context, txSystem TransactionSystem, conf *NodeConf, lo
 		lastLedgerReqTime: time.Time{},
 		log:               log,
 	}
-	n.timer.stop.Store(func() {})
 
 	shardConf, err := n.shardConfStore.GetFirst()
 	if err != nil {

partition/node_init_test.go

@@ -0,0 +1,214 @@
+package partition
+
+import (
+	"context"
+	"crypto"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/unicitynetwork/bft-core/keyvaluedb/memorydb"
+	"github.com/unicitynetwork/bft-core/rootchain/consensus/trustbase"
+	"github.com/unicitynetwork/bft-go-base/types"
+	"github.com/unicitynetwork/bft-go-base/util"
+
+	testcertificates "github.com/unicitynetwork/finality-gadget/internal/testutils/certificates"
+	testlogger "github.com/unicitynetwork/finality-gadget/internal/testutils/logger"
+	testsig "github.com/unicitynetwork/finality-gadget/internal/testutils/sig"
+	testtrustbase "github.com/unicitynetwork/finality-gadget/internal/testutils/trustbase"
+	"github.com/unicitynetwork/finality-gadget/txsystem/state"
+)
+
+type MockTransactionSystem struct {
+	committedUC  *types.UnicityCertificate
+	currentState []byte
+}
+
+func (m *MockTransactionSystem) StateSummary() (*state.Summary, error) {
+	return state.NewStateSummary(m.currentState, []byte{}, 0, nil), nil
+}
+
+func (m *MockTransactionSystem) LeaderPropose(ctx context.Context, round uint64) (*state.Summary, error) {
+	return nil, nil
+}
+
+func (m *MockTransactionSystem) FollowerVerify(ctx context.Context, round uint64, proposedRoot []byte) (*state.Summary, error) {
+	return state.NewStateSummary(proposedRoot, []byte{}, 0, nil), nil
+}
+
+func (m *MockTransactionSystem) UpdateConfig(shardConf *types.PartitionDescriptionRecord) error {
+	return nil
+}
+
+func (m *MockTransactionSystem) RestoreState(ctx context.Context, uc *types.UnicityCertificate) error {
+	m.committedUC = uc
+	m.currentState = uc.GetStateHash()
+	return nil
+}
+
+func (m *MockTransactionSystem) Revert() {}
+
+func (m *MockTransactionSystem) Commit(uc *types.UnicityCertificate) error {
+	m.committedUC = uc
+	m.currentState = uc.GetStateHash()
+	return nil
+}
+
+func (m *MockTransactionSystem) CommittedUC() *types.UnicityCertificate {
+	if m.committedUC == nil {
+		return &types.UnicityCertificate{
+			Version:     1,
+			InputRecord: &types.InputRecord{RoundNumber: 0},
+		}
+	}
+	return m.committedUC
+}
+
+func TestNodeStartupAndRecoveryEquivalence(t *testing.T) {
+	log := testlogger.New(t)
+
+	keyConf, nodeInfo := createKeyConf(t)
+	shardConf := &types.PartitionDescriptionRecord{
+		Version:         1,
+		NetworkID:       3,
+		PartitionID:     3,
+		ShardID:         types.ShardID{},
+		PartitionTypeID: 3,
+		TypeIDLen:       8,
+		UnitIDLen:       256,
+		T2Timeout:       2500 * time.Millisecond,
+		Epoch:           0,
+		EpochStart:      1,
+		Validators:      []*types.NodeInfo{nodeInfo},
+	}
+	signer, _ := testsig.CreateSignerAndVerifier(t)
+	trustBase := testtrustbase.NewTrustBase(t, signer)
+
+	shardDB := memorydb.New()
+	shardConfStore, err := NewShardConfStore(shardDB, log)
+	require.NoError(t, err)
+	require.NoError(t, shardConfStore.Store(shardConf))
+
+	trustBaseStore, err := trustbase.NewTrustBaseStore(memorydb.New(), log)
+	require.NoError(t, err)
+	require.NoError(t, trustBaseStore.Store(trustBase))
+
+	// Create 3 historical blocks
+	var blocks []*types.Block
+	prevBlockHash := []byte("genesis")
+	prevStateHash := []byte(nil)
+	for i := 1; i <= 3; i++ {
+		h := &types.Header{
+			Version:           1,
+			PartitionID:       3,
+			PreviousBlockHash: prevBlockHash,
+			ProposerID:        nodeInfo.NodeID,
+		}
+		txs := make([]*types.TransactionRecord, 0)
+		stateHash := []byte{byte(i)}
+
+		blockHash, err := types.BlockHash(crypto.SHA256, h, []*types.TransactionRecord{}, stateHash, prevStateHash)
+		require.NoError(t, err)
+
+		ir := &types.InputRecord{
+			RoundNumber:  uint64(i),
+			Hash:         stateHash,
+			PreviousHash: prevStateHash,
+			BlockHash:    blockHash,
+			Epoch:        0,
+			SummaryValue: []byte{},
+			ETHash:       []byte{},
+			Timestamp:    uint64(time.Now().UnixMilli()),
+		}
+
+		uc := testcertificates.CreateUnicityCertificate(t, signer, ir, shardConf, uint64(i), nil, make([]byte, 32))
+		ucBytes, err := types.Cbor.Marshal(uc)
+		require.NoError(t, err)
+
+		b := &types.Block{
+			Header:             h,
+			Transactions:       txs,
+			UnicityCertificate: ucBytes,
+		}
+		blocks = append(blocks, b)
+
+		prevBlockHash = uc.GetBlockHash()
+		prevStateHash = uc.GetStateHash()
+	}
+
+	// 1. Simulate Startup Initialization (Fast-forward via blockStore.Last)
+	blockDBInit := memorydb.New()
+	for i, b := range blocks {
+		err := blockDBInit.Write(util.Uint64ToBytes(uint64(i+1)), b)
+		require.NoError(t, err)
+	}
+
+	confInit, err := NewNodeConf(keyConf, shardConfStore, trustBaseStore,
+		WithBlockDB(blockDBInit),
+		WithUnicityCertificateValidator(&AlwaysValidCertificateValidator{}),
+	)
+	require.NoError(t, err)
+
+	txSystemInit := &MockTransactionSystem{currentState: []byte("genesis")}
+	nodeInit := &Node{
+		conf:              confInit,
+		transactionSystem: txSystemInit,
+		blockStore:        confInit.blockDB,
+		state:             &ConsensusState{status: initializing},
+		shardConfStore:    confInit.shardConfStore,
+		trustBaseStore:    confInit.trustBaseStore,
+		log:               log,
+	}
+	err = nodeInit.initState(context.Background())
+	require.NoError(t, err)
+
+	require.Equal(t, uint64(3), txSystemInit.CommittedUC().GetRoundNumber())
+	require.Equal(t, []byte{3}, txSystemInit.currentState)
+
+	// 2. Simulate Network Recovery (Empty DB initially, catch up via handleBlock)
+	blockDBRec := memorydb.New()
+	confRec, err := NewNodeConf(keyConf, shardConfStore, trustBaseStore,
+		WithBlockDB(blockDBRec),
+		WithUnicityCertificateValidator(&AlwaysValidCertificateValidator{}),
+	)
+	require.NoError(t, err)
+
+	// Initial UC (round 0)
+	initialUC := &types.UnicityCertificate{
+		Version: 1,
+		InputRecord: &types.InputRecord{
+			RoundNumber: 0,
+			BlockHash:   []byte("genesis"),
+		},
+	}
+	txSystemRec := &MockTransactionSystem{
+		currentState: []byte(nil),
+		committedUC:  initialUC,
+	}
+
+	nodeRec := &Node{
+		conf:              confRec,
+		transactionSystem: txSystemRec,
+		blockStore:        confRec.blockDB,
+		state:             &ConsensusState{status: recovering},
+		shardConfStore:    confRec.shardConfStore,
+		trustBaseStore:    confRec.trustBaseStore,
+		log:               log,
+	}
+	nodeRec.state.fuc = initialUC
+	nodeRec.state.luc = initialUC
+	nodeRec.shardConf.Store(shardConf)
+
+	// Apply blocks sequentially
+	for _, b := range blocks {
+		err := nodeRec.handleBlock(context.Background(), b)
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, uint64(3), txSystemRec.CommittedUC().GetRoundNumber())
+	require.Equal(t, []byte{3}, txSystemRec.currentState)
+
+	// Compare that the end states of transaction systems are identical
+	require.Equal(t, txSystemInit.CommittedUC().GetRoundNumber(), txSystemRec.CommittedUC().GetRoundNumber())
+	require.Equal(t, txSystemInit.currentState, txSystemRec.currentState)
+}

partition/round.go

@@ -9,25 +9,45 @@ import (
 
 func (n *Node) startNewRound(ctx context.Context) error {
 	n.resetProposal()
-	n.startT1Timer(ctx)
 
 	// not a fatal issue, but log anyway
 	n.deletePendingProposal(ctx)
 	return nil
 }
 
-func (n *Node) startT1Timer(ctx context.Context) {
-	// stop existing timer
-	if stopFunc, ok := n.timer.stop.Load().(func()); ok && stopFunc != nil {
-		stopFunc()
+func (n *Node) ensureT1TimerState(ctx context.Context) {
+	isLeader := n.state.leader == n.peer.ID()
+	shouldBeRunning := isLeader && n.state.status != recovering
+	isRunning := n.timer.isRunning.Load()
+
+	if shouldBeRunning && !isRunning {
+		n.startT1Timer(ctx)
+	} else if !shouldBeRunning && isRunning {
+		n.stopT1Timer()
 	}
+}
+
+func (n *Node) stopT1Timer() {
+	if n.timer.cancel != nil {
+		n.timer.cancel()
+		n.timer.cancel = nil
+	}
+	n.timer.isRunning.Store(false)
+}
+
+func (n *Node) startT1Timer(ctx context.Context) {
+	n.stopT1Timer()
 
 	txCtx, txCancel := context.WithCancel(ctx)
-	n.timer.stop.Store(func() { txCancel() })
+	n.timer.cancel = txCancel
+	n.timer.isRunning.Store(true)
 
 	go func() {
 		select {
 		case <-time.After(n.conf.t1Timeout):
+			// Mark as not running so that ensureT1TimerState can restart the timer
+			n.timer.isRunning.Store(false)
+
 			// Rather than call handleT1TimeoutEvent directly send signal to main
 			// loop - helps to avoid concurrency issues with (repeat) UC handling.
 			select {
@@ -40,11 +60,6 @@ func (n *Node) startT1Timer(ctx context.Context) {
 }
 
 func (n *Node) handleT1TimeoutEvent(ctx context.Context) {
-	if stopFunc, ok := n.timer.stop.Load().(func()); ok && stopFunc != nil {
-		stopFunc()
-	}
-	n.timer.stop.Store(func() {})
-
 	if n.state.status == recovering {
 		n.log.InfoContext(ctx, "T1 timeout: node is recovering")
 		return
@@ -67,10 +82,13 @@ func (n *Node) handleT1TimeoutEvent(ctx context.Context) {
 
 	if err := n.sendBlockProposal(ctx, stateSummary.Root()); err != nil {
 		n.log.WarnContext(ctx, "Failed to send BlockProposal", logger.Error(err))
+		n.transactionSystem.Revert()
 		return
 	}
 
 	if err := n.sendCertificationRequest(ctx, n.peer.ID().String(), stateSummary); err != nil {
 		n.log.WarnContext(ctx, "Failed to send certification request", logger.Error(err))
+		n.transactionSystem.Revert()
+		return
 	}
 }