The real v2 - new database and new data model + gRPC

blindbit.example.toml

@@ -3,15 +3,24 @@ log_level = "debug"
 
 # 0.0.0.0:8000 to expose outside of localhost
 # default: "127.0.0.1:8000"
-host = "127.0.0.1:8000"
+http_host = "127.0.0.1:8000"
+
+# defines where the gRPC host is exposed
+# default: "" gRPC is not exposed by default
+grpc_host = "127.0.0.1:8001"
 
 # Defines on which chain the wallet runs. Allowed values: main, testnet, signet, regtest.
 # default: signet
 chain = "signet"
 
+# bitcoin core rpc endpoint
 # default: http://127.0.0.1:8332
-# also used for rest api
-rpc_endpoint = "http://127.0.0.1:18443"
+core_rpc_endpoint = "http://127.0.0.1:18443"
+
+# bitcoin core rest endpoint
+# note: requires bitcoin core PR #32540 (already merged to master but in a release yet)
+# default: ""
+core_rest_endpoint = "http://127.0.01:38332"
 
 # required, unless rpc_user and rpc_pass are set
 cookie_path = ""

cmd/blindbit-oracle/main.go

@@ -58,7 +58,10 @@ func init() {
 		return
 	}
 
-	config.SetDirectories() // todo a proper set settings function which does it all would be good to avoid several small function calls
+	// todo: a proper set settings function which does it all
+	// avoid several small function calls
+	config.SetDirectories()
+
 	err := os.Mkdir(config.BaseDirectory, 0750)
 	if err != nil && !errors.Is(err, os.ErrExist) {
 		logging.L.Fatal().Err(err).Msg("error creating base directory")
@@ -75,9 +78,6 @@ func init() {
 		logging.L.Fatal().Err(err).Msg("error creating db path")
 	}
 
-	// open levelDB connections
-	// openLevelDBConnections()
-
 	if config.CookiePath != "" {
 		data, err := os.ReadFile(config.CookiePath)
 		if err != nil {
@@ -147,24 +147,19 @@ func main() {
 
 	// index builder
 	go func() {
-		// err = database.DropIndexesForIBD(context.Background(), db)
-		// if err != nil {
-		// 	logging.L.Err(err).Msg("failed to drop indexes")
-		// 	errChan <- err
-		// }
-
 		builder := indexer.NewBuilder(store)
-		// builder := indexer.NewBuilder(db)
 
-		// ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
-		// defer cancel()
+		// do initial sync then move towards steady state sync
+		err = builder.InitialSyncToTip(ctx)
+		if err != nil {
+			logging.L.Err(err).Msg("failed initial sync")
+			errChan <- err
+			return
+		}
 
-		_ = ctx
-		_ = builder
+		// do continous scans
+		err = builder.ContinuousSync(ctx)
 
-		// err = builder.SyncBlocks(ctx, 1, 265_963)
-		// err = builder.SyncBlocks(ctx, 251_000, 265_000)
-		// err = builder.SyncBlocks(ctx, 230_000, 265_000)
 		if err != nil {
 			logging.L.Err(err).Msg("error indexing blocks")
 			errChan <- err

internal/config/config.go

@@ -38,7 +38,8 @@ func LoadConfigs(pathToConfig string) {
 	viper.BindEnv("http_host", "HTTP_HOST")
 	viper.BindEnv("grpc_host", "GRPC_HOST")
 	viper.BindEnv("chain", "CHAIN")
-	viper.BindEnv("rpc_endpoint", "RPC_ENDPOINT")
+	viper.BindEnv("core_rpc_endpoint", "CORE_RPC_ENDPOINT")
+	viper.BindEnv("core_rest_endpoint", "CORE_REST_ENDPOINT")
 	viper.BindEnv("cookie_path", "COOKIE_PATH")
 	viper.BindEnv("rpc_pass", "RPC_PASS")
 	viper.BindEnv("rpc_user", "RPC_USER")

internal/config/vars.go

@@ -28,10 +28,11 @@ var (
 )
 
 var (
-	RpcEndpoint = "http://127.0.0.1:8332" // default local node
-	CookiePath  = ""
-	RpcUser     = ""
-	RpcPass     = ""
+	RpcEndpoint  = "http://127.0.0.1:8332" // default local node
+	RestEndpoint = ""                      // default local node
+	CookiePath   = ""
+	RpcUser      = ""
+	RpcPass      = ""
 
 	BaseDirectory = ""
 	DBPath        = ""

internal/database/dbpebble/read.go

@@ -21,8 +21,8 @@ func (s *Store) GetChainTip() ([]byte, uint32, error) {
 	}
 	defer it.Close()
 	if !it.Last() {
-		// means literally zero blockhashes are in the db
-		return nil, 0, errors.New("no chain tip found")
+		// edge case empty db we are at 0 height
+		return nil, 0, nil
 	}
 	heightBytes := it.Key()
 	blockhash := it.Value()

internal/database/dbpebble/store.go

@@ -63,7 +63,9 @@ func (s *Store) collectAndWrite(block *database.DBBlock) error {
 			logging.L.Err(err).Msg("failed to write Batch")
 			return err
 		}
-		logging.L.Warn().Dur("write_batch_duration", time.Since(writeBatchStart)).Msg("batch_write_bench")
+		logging.L.Warn().
+			Dur("write_batch_duration", time.Since(writeBatchStart)).
+			Msg("batch_write_bench")
 		err = s.dbBatch.Close()
 		if err != nil {
 			logging.L.Err(err).Msg("failed to close db batch")
@@ -83,6 +85,31 @@ func (s *Store) collectAndWrite(block *database.DBBlock) error {
 	return nil
 }
 
+func (s *Store) FlushBatch() error {
+	s.batchSync.Lock()
+	defer s.batchSync.Unlock()
+
+	err := s.dbBatch.Commit(pebble.NoSync)
+	if err != nil {
+		logging.L.Err(err).Msg("failed to write Batch")
+		return err
+	}
+	return nil
+}
+
+// don't use yet until api design is clear for this concept
+func (s *Store) commitBatch() error {
+	s.batchSync.Lock()
+	defer s.batchSync.Unlock()
+
+	err := s.dbBatch.Commit(pebble.NoSync)
+	if err != nil {
+		logging.L.Err(err).Msg("failed to write Batch")
+		return err
+	}
+	return nil
+}
+
 func attachBlcokToBatch(batch *pebble.Batch, block *database.DBBlock) error {
 	blockHash := block.Hash[:]
 	txs := block.Txs
@@ -103,7 +130,14 @@ func attachBlcokToBatch(batch *pebble.Batch, block *database.DBBlock) error {
 	}
 
 	// block → txs + transaction records
-	for i, t := range txs {
+	for i := range txs {
+
+		t := txs[i]
+		if t == nil {
+			// we skip nil txs here to so that we have exact positions.
+			// todo: should we keep it like this?
+			continue
+		}
 		// bt
 		if err := b.Set(KeyBlockTx(blockHash, uint32(i)), t.Txid, nil); err != nil {
 			logging.L.Err(err).Msg("insert failed")

internal/database/specifications.go

@@ -8,6 +8,7 @@ type DB interface {
 	GetChainTip() ([]byte, uint32, error)
 	GetBlockHashByHeight(height uint32) ([]byte, error)
 	ApplyBlock(*DBBlock) error
+	FlushBatch() error
 	TweaksForBlockAll([]byte) ([]TweakRow, error)
 	TweaksForBlockCutThrough([]byte, uint32) ([]TweakRow, error)
 }