more testcase fixes

go.mod

@@ -179,7 +179,7 @@ require (
 	go.uber.org/fx v1.23.0 // indirect
 	go.uber.org/mock v0.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.0
+	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.29.0 // indirect
 	golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f // indirect
 	golang.org/x/mod v0.22.0 // indirect

node/consensus/engine_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/kwilteam/kwil-db/node/meta"
 	"github.com/kwilteam/kwil-db/node/pg"
 	dbtest "github.com/kwilteam/kwil-db/node/pg/test"
+	"github.com/kwilteam/kwil-db/node/snapshotter"
 	"github.com/kwilteam/kwil-db/node/store"
 	"github.com/kwilteam/kwil-db/node/txapp"
 	"github.com/kwilteam/kwil-db/node/types"
@@ -95,6 +96,8 @@ func generateTestCEConfig(t *testing.T, nodes int, leaderDB bool) []*Config {
 		require.NoError(t, tx.Commit(ctx))
 	}()
 
+	ss := &snapshotStore{}
+
 	for i := range nodes {
 		nodeStr := fmt.Sprintf("NODE%d", i)
 		nodeDir := filepath.Join(tempDir, nodeStr)
@@ -113,6 +116,7 @@ func generateTestCEConfig(t *testing.T, nodes int, leaderDB bool) []*Config {
 			TxApp:          txapp,
 			Accounts:       accounts,
 			ValidatorSet:   validatorSet,
+			Snapshots:      ss,
 			ValidatorStore: v,
 			Logger:         logger,
 			ProposeTimeout: 1 * time.Second,
@@ -596,7 +600,7 @@ func TestCELeaderSingleNode(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		return leader.lastCommitHeight() >= 1 // Ensure that the leader mines a block
-	}, 6*time.Second, 100*time.Millisecond)
+	}, 10*time.Second, 100*time.Millisecond)
 }
 
 func TestCELeaderTwoNodesMajorityAcks(t *testing.T) {
@@ -812,3 +816,27 @@ func (ce *ConsensusEngine) blockResult() *blockResult {
 
 	return ce.state.blockRes
 }
+
+// should satisfy the SnapshotModule interface
+type snapshotStore struct {
+}
+
+func (s *snapshotStore) Enabled() bool {
+	return false
+}
+
+func (s *snapshotStore) ListSnapshots() []*snapshotter.Snapshot {
+	return nil
+}
+
+func (s *snapshotStore) CreateSnapshot(ctx context.Context, height uint64, snapshotID string, schemas, excludedTables []string, excludeTableData []string) error {
+	return nil
+}
+
+func (s *snapshotStore) IsSnapshotDue(height uint64) bool {
+	return false
+}
+
+func (s *snapshotStore) LoadSnapshotChunk(height uint64, format uint32, chunkID uint32) ([]byte, error) {
+	return nil, nil
+}

node/node_live_test.go

@@ -86,6 +86,7 @@ func TestDualNodeMocknet(t *testing.T) {
 	for _, v := range valSet {
 		valSetList = append(valSetList, &v)
 	}
+	ss := newSnapshotStore()
 
 	ceCfg1 := &consensus.Config{
 		PrivateKey:     privKeys[0],
@@ -99,19 +100,22 @@ func TestDualNodeMocknet(t *testing.T) {
 		Logger:         log.New(log.WithName("CE1"), log.WithWriter(os.Stdout), log.WithLevel(log.LevelDebug), log.WithFormat(log.FormatUnstructured)),
 		ProposeTimeout: 1 * time.Second,
 		DB:             db1,
+		Snapshots:      ss,
 	}
 	ce1 := consensus.New(ceCfg1)
 	defaultConfigSet := config.DefaultConfig()
-
 	log1 := log.New(log.WithName("NODE1"), log.WithWriter(os.Stdout), log.WithLevel(log.LevelDebug), log.WithFormat(log.FormatUnstructured))
 	cfg1 := &Config{
-		RootDir:    root1,
-		PrivKey:    privKeys[0],
-		Logger:     log1,
-		P2P:        &defaultConfigSet.P2P,
-		Mempool:    mp1,
-		BlockStore: bs1,
-		Consensus:  ce1,
+		RootDir:     root1,
+		PrivKey:     privKeys[0],
+		Logger:      log1,
+		P2P:         &defaultConfigSet.P2P,
+		Mempool:     mp1,
+		BlockStore:  bs1,
+		Consensus:   ce1,
+		Snapshotter: ss,
+		DBConfig:    &defaultConfigSet.DB,
+		Statesync:   &defaultConfigSet.StateSync,
 	}
 	node1, err := NewNode(cfg1, WithHost(h1))
 	if err != nil {
@@ -139,18 +143,22 @@ func TestDualNodeMocknet(t *testing.T) {
 		Logger:         log.New(log.WithName("CE2"), log.WithWriter(os.Stdout), log.WithLevel(log.LevelDebug), log.WithFormat(log.FormatUnstructured)),
 		ProposeTimeout: 1 * time.Second,
 		DB:             db2,
+		Snapshots:      ss,
 	}
 	ce2 := consensus.New(ceCfg2)
 
 	log2 := log.New(log.WithName("NODE2"), log.WithWriter(os.Stdout), log.WithLevel(log.LevelDebug), log.WithFormat(log.FormatUnstructured))
 	cfg2 := &Config{
-		RootDir:    root2,
-		PrivKey:    privKeys[1],
-		Logger:     log2,
-		P2P:        &defaultConfigSet.P2P,
-		Mempool:    mp2,
-		BlockStore: bs2,
-		Consensus:  ce2,
+		RootDir:     root2,
+		PrivKey:     privKeys[1],
+		Logger:      log2,
+		P2P:         &defaultConfigSet.P2P,
+		Mempool:     mp2,
+		BlockStore:  bs2,
+		Consensus:   ce2,
+		Snapshotter: ss,
+		DBConfig:    &defaultConfigSet.DB,
+		Statesync:   &defaultConfigSet.StateSync,
 	}
 	node2, err := NewNode(cfg2, WithHost(h2))
 	if err != nil {
@@ -192,7 +200,7 @@ func TestDualNodeMocknet(t *testing.T) {
 		h, _, _, err := meta.GetChainState(ctx, tx)
 		require.NoError(t, err)
 		assert.GreaterOrEqual(c, h, reachHeight)
-	}, 6*time.Second, 250*time.Millisecond)
+	}, 10*time.Second, 250*time.Millisecond)
 
 	cancel()
 	wg.Wait()

node/node_test.go

@@ -287,7 +287,7 @@ func TestStreamsBlockFetch(t *testing.T) {
 	defaultConfigSet := config.DefaultConfig()
 	defaultConfigSet.Consensus.ProposeTimeout = 5 * time.Minute
 
-	ss := NewSnapshotStore()
+	ss := newSnapshotStore()
 
 	// log1 := log.New(log.WithName("NODE1"), log.WithWriter(os.Stdout), log.WithLevel(log.LevelDebug), log.WithFormat(log.FormatUnstructured))
 	cfg1 := &Config{

node/statesync.go

@@ -73,7 +73,7 @@ type StateSyncService struct {
 }
 
 func NewStateSyncService(ctx context.Context, cfg *statesyncConfig) (*StateSyncService, error) {
-	if cfg.StateSyncCfg.Enable && (cfg.StateSyncCfg.TrustedProviders == nil || len(cfg.StateSyncCfg.TrustedProviders) == 0) {
+	if cfg.StateSyncCfg.Enable && cfg.StateSyncCfg.TrustedProviders == nil {
 		return nil, fmt.Errorf("at least one trusted provider is required for state sync")
 	}
 

node/statesync_test.go

@@ -41,22 +41,20 @@ var (
 		Size:        100,
 		ChunkHashes: [][32]byte{data},
 	}
-
-	invalidSnap1 = &snapshotMetadata{
-		Height:      1,
-		Format:      1,
-		Chunks:      1,
-		Hash:        []byte("snap1-invalid"),
-		Size:        100,
-		ChunkHashes: [][32]byte{data},
-	}
 )
 
+type mockBS struct {
+}
+
+func (m *mockBS) GetByHeight(height int64) (types.Hash, *types.Block, types.Hash, error) {
+	return types.Hash{}, nil, types.Hash{}, nil
+}
+
 type snapshotStore struct {
 	snapshots map[uint64]*snapshotMetadata
 }
 
-func NewSnapshotStore() *snapshotStore {
+func newSnapshotStore() *snapshotStore {
 	return &snapshotStore{
 		snapshots: make(map[uint64]*snapshotMetadata),
 	}
@@ -66,13 +64,6 @@ func (s *snapshotStore) addSnapshot(snapshot *snapshotMetadata) {
 	s.snapshots[snapshot.Height] = snapshot
 }
 
-type mockBS struct {
-}
-
-func (m *mockBS) GetByHeight(height int64) (types.Hash, *types.Block, types.Hash, error) {
-	return types.Hash{}, nil, types.Hash{}, nil
-}
-
 func (s *snapshotStore) ListSnapshots() []*snapshotter.Snapshot {
 	snapshots := make([]*snapshotter.Snapshot, 0, len(s.snapshots))
 	for _, snapshot := range s.snapshots {
@@ -127,6 +118,14 @@ func (s *snapshotStore) Enabled() bool {
 	return true
 }
 
+func (s *snapshotStore) IsSnapshotDue(height uint64) bool {
+	return false
+}
+
+func (s *snapshotStore) CreateSnapshot(ctx context.Context, height uint64, snapshotID string, schemas, excludedTables []string, excludeTableData []string) error {
+	return nil
+}
+
 func newTestStatesyncer(ctx context.Context, t *testing.T, mn mock.Mocknet, rootDir string, sCfg *config.StateSyncConfig) (host.Host, discovery.Discovery, *snapshotStore, *StateSyncService, error) {
 	_, h, err := newTestHost(t, mn)
 	if err != nil {
@@ -143,7 +142,7 @@ func newTestStatesyncer(ctx context.Context, t *testing.T, mn mock.Mocknet, root
 	os.MkdirAll(rootDir, os.ModePerm)
 
 	bs := &mockBS{}
-	st := NewSnapshotStore()
+	st := newSnapshotStore()
 	cfg := &statesyncConfig{
 		StateSyncCfg: sCfg,
 		RcvdSnapsDir: rootDir,