feat: initial ledger state database

config.go

@@ -23,6 +23,7 @@ import (
 )
 
 type Config struct {
+	dataDir            string
 	logger             *slog.Logger
 	listeners          []ListenerConfig
 	network            string
@@ -90,6 +91,13 @@ func NewConfig(opts ...ConfigOptionFunc) Config {
 	return c
 }
 
+// WithDataDir specifies the persistent data directory to use. The default is to store everything in memory
+func WithDataDir(dataDir string) ConfigOptionFunc {
+	return func(c *Config) {
+		c.dataDir = dataDir
+	}
+}
+
 // WithLogger specifies the logger to use. This defaults to discarding log output
 func WithLogger(logger *slog.Logger) ConfigOptionFunc {
 	return func(c *Config) {

database/database.go

@@ -0,0 +1,171 @@
+// Copyright 2024 Blink Labs Software
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package database
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"time"
+
+	badger "github.com/dgraph-io/badger/v4"
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+)
+
+type Database interface {
+	Metadata() *gorm.DB
+	Blob() *badger.DB
+}
+
+type BaseDatabase struct {
+	logger      *slog.Logger
+	metadata    *gorm.DB
+	blob        *badger.DB
+	blobGcTimer *time.Ticker
+}
+
+// Metadata returns the underlying metadata DB instance
+func (b *BaseDatabase) Metadata() *gorm.DB {
+	return b.metadata
+}
+
+// Blob returns the underling blob DB instance
+func (b *BaseDatabase) Blob() *badger.DB {
+	return b.blob
+}
+
+func (b *BaseDatabase) init() error {
+	if b.logger == nil {
+		// Create logger to throw away logs
+		// We do this so we don't have to add guards around every log operation
+		b.logger = slog.New(slog.NewJSONHandler(io.Discard, nil))
+	}
+	// Run GC periodically for Badger DB
+	b.blobGcTimer = time.NewTicker(5 * time.Minute)
+	go b.blobGc()
+	return nil
+}
+
+func (b *BaseDatabase) blobGc() {
+	for range b.blobGcTimer.C {
+	again:
+		err := b.blob.RunValueLogGC(0.5)
+		if err != nil {
+			// Log any actual errors
+			if !errors.Is(err, badger.ErrNoRewrite) {
+				b.logger.Warn(
+					fmt.Sprintf("blob DB: GC failure: %s", err),
+				)
+			}
+		} else {
+			// Run it again if it just ran successfully
+			goto again
+		}
+	}
+}
+
+// InMemoryDatabase stores all data in memory. Data will not be persisted
+type InMemoryDatabase struct {
+	*BaseDatabase
+}
+
+// NewInMemory creates a new in-memory database
+func NewInMemory(logger *slog.Logger) (*InMemoryDatabase, error) {
+	// Open sqlite DB
+	metadataDb, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		return nil, err
+	}
+	// Open Badger DB
+	badgerOpts := badger.DefaultOptions("").
+		WithLogger(NewBadgerLogger(logger)).
+		// The default INFO logging is a bit verbose
+		WithLoggingLevel(badger.WARNING).
+		WithInMemory(true)
+	blobDb, err := badger.Open(badgerOpts)
+	if err != nil {
+		return nil, err
+	}
+	db := &InMemoryDatabase{
+		BaseDatabase: &BaseDatabase{
+			logger:   logger,
+			metadata: metadataDb,
+			blob:     blobDb,
+		},
+	}
+	if err := db.init(); err != nil {
+		return nil, err
+	}
+	return db, nil
+}
+
+// PersistentDatabase stores its data on disk, providing persistence across restarts
+type PersistentDatabase struct {
+	*BaseDatabase
+	dataDir string
+}
+
+// NewPersistent creates a new persistent database instance using the provided data directory
+func NewPersistent(dataDir string, logger *slog.Logger) (*PersistentDatabase, error) {
+	// Make sure that we can read data dir, and create if it doesn't exist
+	if _, err := os.Stat(dataDir); err != nil {
+		if !errors.Is(err, fs.ErrNotExist) {
+			return nil, fmt.Errorf("failed to read data dir: %w", err)
+		}
+		// Create data directory
+		if err := os.MkdirAll(dataDir, fs.ModePerm); err != nil {
+			return nil, fmt.Errorf("failed to create data dir: %w", err)
+		}
+	}
+	// Open sqlite DB
+	metadataDbPath := filepath.Join(
+		dataDir,
+		"metadata.sqlite",
+	)
+	metadataDb, err := gorm.Open(sqlite.Open(metadataDbPath), &gorm.Config{})
+	if err != nil {
+		return nil, err
+	}
+	// Open Badger DB
+	blobDir := filepath.Join(
+		dataDir,
+		"blob",
+	)
+	badgerOpts := badger.DefaultOptions(blobDir).
+		WithLogger(NewBadgerLogger(logger)).
+		// The default INFO logging is a bit verbose
+		WithLoggingLevel(badger.WARNING)
+	blobDb, err := badger.Open(badgerOpts)
+	if err != nil {
+		return nil, err
+	}
+	db := &PersistentDatabase{
+		BaseDatabase: &BaseDatabase{
+			logger:   logger,
+			metadata: metadataDb,
+			blob:     blobDb,
+		},
+		dataDir: dataDir,
+	}
+	if err := db.init(); err != nil {
+		return nil, err
+	}
+	return db, nil
+}

database/logger.go

@@ -0,0 +1,59 @@
+// Copyright 2024 Blink Labs Software
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package database
+
+import (
+	"fmt"
+	"io"
+	"log/slog"
+)
+
+// BadgerLogger is a wrapper type to give our logger the expected interface
+type BadgerLogger struct {
+	logger *slog.Logger
+}
+
+func NewBadgerLogger(logger *slog.Logger) *BadgerLogger {
+	if logger == nil {
+		// Create logger to throw away logs
+		// We do this so we don't have to add guards around every log operation
+		logger = slog.New(slog.NewJSONHandler(io.Discard, nil))
+	}
+	return &BadgerLogger{logger: logger}
+}
+
+func (b *BadgerLogger) Infof(msg string, args ...any) {
+	b.logger.Info(
+		fmt.Sprintf(msg, args...),
+	)
+}
+
+func (b *BadgerLogger) Warningf(msg string, args ...any) {
+	b.logger.Warn(
+		fmt.Sprintf(msg, args...),
+	)
+}
+
+func (b *BadgerLogger) Debugf(msg string, args ...any) {
+	b.logger.Debug(
+		fmt.Sprintf(msg, args...),
+	)
+}
+
+func (b *BadgerLogger) Errorf(msg string, args ...any) {
+	b.logger.Error(
+		fmt.Sprintf(msg, args...),
+	)
+}

go.mod

@@ -4,33 +4,52 @@ go 1.21.5
 
 require (
 	github.com/blinklabs-io/gouroboros v0.91.1
+	github.com/dgraph-io/badger/v4 v4.2.0
+	github.com/glebarez/sqlite v1.11.0
 	github.com/prometheus/client_golang v1.19.1
 	github.com/spf13/cobra v1.8.1
 	go.opentelemetry.io/otel v1.28.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.28.0
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.28.0
 	go.opentelemetry.io/otel/sdk v1.28.0
 	golang.org/x/sys v0.23.0
+	gorm.io/gorm v1.25.11
 )
 
 require (
 	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/dgraph-io/ristretto v0.1.1 // indirect
+	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
+	github.com/glebarez/go-sqlite v1.21.2 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/glog v1.2.0 // indirect
+	github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/golang/snappy v0.0.3 // indirect
+	github.com/google/flatbuffers v1.12.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jinzhu/copier v0.4.0 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/klauspost/compress v1.12.3 // indirect
+	github.com/mattn/go-isatty v0.0.17 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.52.2 // indirect
 	github.com/prometheus/procfs v0.13.0 // indirect
+	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/utxorpc/go-codegen v0.8.0 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	go.opencensus.io v0.22.5 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.28.0 // indirect
 	go.opentelemetry.io/otel/metric v1.28.0 // indirect
 	go.opentelemetry.io/otel/trace v1.28.0 // indirect
@@ -42,4 +61,8 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240701130421-f6361c86f094 // indirect
 	google.golang.org/grpc v1.64.1 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
+	modernc.org/libc v1.22.5 // indirect
+	modernc.org/mathutil v1.5.0 // indirect
+	modernc.org/memory v1.5.0 // indirect
+	modernc.org/sqlite v1.23.1 // indirect
 )

node.go

@@ -23,6 +23,7 @@ import (
 	"github.com/blinklabs-io/node/chainsync"
 	"github.com/blinklabs-io/node/event"
 	"github.com/blinklabs-io/node/mempool"
+	"github.com/blinklabs-io/node/state"
 
 	ouroboros "github.com/blinklabs-io/gouroboros"
 )
@@ -35,6 +36,7 @@ type Node struct {
 	outboundConns      map[ouroboros.ConnectionId]outboundPeer
 	outboundConnsMutex sync.Mutex
 	mempool            *mempool.Mempool
+	ledgerState        *state.LedgerState
 	shutdownFuncs      []func(context.Context) error
 }
 
@@ -63,6 +65,15 @@ func (n *Node) Run() error {
 			return err
 		}
 	}
+	// Load state
+	state, err := state.NewLedgerState(
+		n.config.dataDir,
+		n.config.logger,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to load state database: %w", err)
+	}
+	n.ledgerState = state
 	// Configure connection manager
 	n.connManager = ouroboros.NewConnectionManager(
 		ouroboros.ConnectionManagerConfig{

state/models/block.go

@@ -0,0 +1,24 @@
+// Copyright 2024 Blink Labs Software
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package models
+
+type Block struct {
+	ID     uint `gorm:"primarykey"`
+	Slot   uint64
+	Number uint64
+	Hash   []byte
+	Type   uint8
+	Cbor   []byte `gorm:"-"` // This is here for convenience but not represented in the metadata DB
+}

state/state.go

@@ -0,0 +1,121 @@
+// Copyright 2024 Blink Labs Software
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package state
+
+import (
+	"errors"
+	"io"
+	"log/slog"
+	"math/big"
+
+	"github.com/blinklabs-io/node/database"
+	"github.com/blinklabs-io/node/state/models"
+	"gorm.io/gorm"
+
+	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
+	badger "github.com/dgraph-io/badger/v4"
+)
+
+type LedgerState struct {
+	logger  *slog.Logger
+	dataDir string
+	db      database.Database
+}
+
+func NewLedgerState(dataDir string, logger *slog.Logger) (*LedgerState, error) {
+	ls := &LedgerState{
+		dataDir: dataDir,
+		logger:  logger,
+	}
+	if logger == nil {
+		// Create logger to throw away logs
+		// We do this so we don't have to add guards around every log operation
+		ls.logger = slog.New(slog.NewJSONHandler(io.Discard, nil))
+	}
+	if dataDir == "" {
+		db, err := database.NewInMemory(ls.logger)
+		if err != nil {
+			return nil, err
+		}
+		ls.db = db
+	} else {
+		db, err := database.NewPersistent(dataDir, ls.logger)
+		if err != nil {
+			return nil, err
+		}
+		ls.db = db
+	}
+	// Create the table schemas
+	if err := ls.db.Metadata().AutoMigrate(&models.Block{}); err != nil {
+		return nil, err
+	}
+	return ls, nil
+}
+
+func (ls *LedgerState) AddBlock(block models.Block) error {
+	// Add block to blob DB
+	slotBytes := make([]byte, 8)
+	new(big.Int).SetUint64(block.Slot).FillBytes(slotBytes)
+	key := []byte("b")
+	key = append(key, slotBytes...)
+	key = append(key, block.Hash...)
+	err := ls.db.Blob().Update(func(txn *badger.Txn) error {
+		err := txn.Set(key, block.Cbor)
+		return err
+	})
+	if err != nil {
+		return err
+	}
+	// Add to metadata DB
+	if result := ls.db.Metadata().Create(&block); result.Error != nil {
+		return result.Error
+	}
+	return nil
+}
+
+func (ls *LedgerState) GetBlock(point ocommon.Point) (*models.Block, error) {
+	var ret models.Block
+	// Block metadata
+	result := ls.db.Metadata().First(&ret, "slot = ? AND hash = ?", point.Slot, point.Hash)
+	if result.Error != nil {
+		if !errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, result.Error
+		}
+		return nil, nil
+	}
+	// Block CBOR
+	slotBytes := make([]byte, 8)
+	new(big.Int).SetUint64(point.Slot).FillBytes(slotBytes)
+	key := []byte("b")
+	key = append(key, slotBytes...)
+	key = append(key, point.Hash...)
+	err := ls.db.Blob().View(func(txn *badger.Txn) error {
+		item, err := txn.Get(key)
+		if err != nil {
+			return err
+		}
+		ret.Cbor, err = item.ValueCopy(nil)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		if errors.Is(err, badger.ErrKeyNotFound) {
+			return nil, nil
+		}
+	}
+	return &ret, nil
+}