luxfi/node
1
0
Fork 0
mirror of https://github.com/luxfi/node.git synced 2026-07-27 03:39:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

rpcdb: consolidate into node/db/rpcdb, drop luxfi/proto dep (#109)

Browse Source 
Part A — swap Layer-B wire-types path:
  github.com/luxfi/proto/rpcdb → github.com/luxfi/protocol/rpcdb (v0.0.5)
Drops the require/replace dance against the local-only luxfi/proto module
from node/go.mod. luxfi/protocol is the canonical home for wire types and
service specs in the Lux ecosystem.

Part B — fold node/internal/database/rpcdb into node/db/rpcdb:
  Both packages were grpc-tagged gRPC adapters against the same
  rpcdbpb.DatabaseClient/Server. The internal one (db_client.go/db_server.go)
  predated the Layer A/B/C decomplect; the db/rpcdb one (grpc_server.go +
  zap_server.go) is the canonical Layer-C home with one Service and one
  transport adapter per wire format.

  New grpc_client.go in db/rpcdb mirrors the deleted internal db_client.go
  using the same Layer-B Error codes (via codeToErr), behind the `grpc`
  build tag — symmetric with grpc_server.go.

  Updated service/keystore/rpckeystore/client.go to import the canonical
  db/rpcdb.NewGRPCClient instead of internal/database/rpcdb.NewClient.

  internal/database/rpcdb/ deleted in its entirety. One canonical rpcdb
  home; no dual-shim, no deprecation period.

Default-tag build is clean. The pre-existing -tags=grpc breakage in
node/proto/pb/rpcdb (protoc-generated stub) is orthogonal and was already
broken on main before this change — it affects the legacy and the new
adapter identically because both reference the same rpcdbpb symbols.
This commit is contained in:
Hanzo Dev
2026-05-16 16:35:56 -07:00
committed by GitHub
parent 4ea2f95be8
commit 8aa002d245
11 changed files with 407 additions and 647 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/rpcdb/grpc_client.go
+370
View File
@@ -0,0 +1,370 @@
//go:build grpc
// Copyright (C) 2019-2026, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package rpcdb
import (
"context"
"encoding/json"
"errors"
"io"
"sync"
"google.golang.org/protobuf/types/known/emptypb"
"github.com/luxfi/database"
"github.com/luxfi/math/set"
"github.com/luxfi/utils"
rpcdbpb "github.com/luxfi/node/proto/pb/rpcdb"
)
var (
_ database.Database = (*GRPCClient)(nil)
_ database.Batch = (*grpcBatch)(nil)
_ database.Iterator = (*grpcIterator)(nil)
)
// GRPCClient is a database.Database that talks to a GRPCServer over
// gRPC. Symmetric with GRPCServer — both are pure transport adapters
// against the Layer-B wire types in github.com/luxfi/protocol/rpcdb.
type GRPCClient struct {
client rpcdbpb.DatabaseClient
closed utils.Atomic[bool]
}
// NewGRPCClient wraps a gRPC client connection to a remote rpcdb
// service.
func NewGRPCClient(client rpcdbpb.DatabaseClient) *GRPCClient {
return &GRPCClient{client: client}
}
// Has attempts to return if the database has a key with the provided value.
func (c *GRPCClient) Has(key []byte) (bool, error) {
resp, err := c.client.Has(context.Background(), &rpcdbpb.HasRequest{Key: key})
if err != nil {
return false, err
}
return resp.Has, codeToErr(resp.Err)
}
// Get attempts to return the value mapped to the key.
func (c *GRPCClient) Get(key []byte) ([]byte, error) {
resp, err := c.client.Get(context.Background(), &rpcdbpb.GetRequest{Key: key})
if err != nil {
return nil, err
}
return resp.Value, codeToErr(resp.Err)
}
// Put attempts to set the value this key maps to.
func (c *GRPCClient) Put(key, value []byte) error {
resp, err := c.client.Put(context.Background(), &rpcdbpb.PutRequest{
Key: key,
Value: value,
})
if err != nil {
return err
}
return codeToErr(resp.Err)
}
// Delete attempts to remove any mapping from the key.
func (c *GRPCClient) Delete(key []byte) error {
resp, err := c.client.Delete(context.Background(), &rpcdbpb.DeleteRequest{Key: key})
if err != nil {
return err
}
return codeToErr(resp.Err)
}
// NewBatch returns a new batch.
func (c *GRPCClient) NewBatch() database.Batch {
return &grpcBatch{db: c}
}
func (c *GRPCClient) NewIterator() database.Iterator {
return c.NewIteratorWithStartAndPrefix(nil, nil)
}
func (c *GRPCClient) NewIteratorWithStart(start []byte) database.Iterator {
return c.NewIteratorWithStartAndPrefix(start, nil)
}
func (c *GRPCClient) NewIteratorWithPrefix(prefix []byte) database.Iterator {
return c.NewIteratorWithStartAndPrefix(nil, prefix)
}
// NewIteratorWithStartAndPrefix returns a new iterator.
func (c *GRPCClient) NewIteratorWithStartAndPrefix(start, prefix []byte) database.Iterator {
resp, err := c.client.NewIteratorWithStartAndPrefix(
context.Background(),
&rpcdbpb.NewIteratorWithStartAndPrefixRequest{Start: start, Prefix: prefix},
)
if err != nil {
return &database.IteratorError{Err: err}
}
return newGRPCIterator(c, resp.Id)
}
// Compact attempts to optimize the space utilization in the provided range.
func (c *GRPCClient) Compact(start, limit []byte) error {
resp, err := c.client.Compact(context.Background(), &rpcdbpb.CompactRequest{
Start: start,
Limit: limit,
})
if err != nil {
return err
}
return codeToErr(resp.Err)
}
// Close attempts to close the database.
func (c *GRPCClient) Close() error {
c.closed.Set(true)
resp, err := c.client.Close(context.Background(), &rpcdbpb.CloseRequest{})
if err != nil {
return err
}
return codeToErr(resp.Err)
}
// Sync — rpc database delegates sync to the underlying database on the
// server side. Operations are already synchronous over RPC; no-op here
// matches the legacy internal/database/rpcdb behavior.
func (c *GRPCClient) Sync() error {
return nil
}
func (c *GRPCClient) HealthCheck(ctx context.Context) (interface{}, error) {
health, err := c.client.HealthCheck(ctx, &emptypb.Empty{})
if err != nil {
return nil, err
}
return json.RawMessage(health.Details), nil
}
// Backup is not supported over the gRPC database client.
func (c *GRPCClient) Backup(_ io.Writer, _ uint64) (uint64, error) {
return 0, errors.New("rpcdb: backup not supported")
}
// Load is not supported over the gRPC database client.
func (c *GRPCClient) Load(_ io.Reader) error {
return errors.New("rpcdb: load not supported")
}
// codeToErr maps a generated-pb Error enum back to a database sentinel.
// Lives next to the gRPC adapter so the adapter knows the inverse of
// toPbErr. Mirrors rpcdb.CodeToErr in service.go but for the
// protobuf-typed enum.
func codeToErr(code rpcdbpb.Error) error {
switch code {
case rpcdbpb.Error_ERROR_CLOSED:
return database.ErrClosed
case rpcdbpb.Error_ERROR_NOT_FOUND:
return database.ErrNotFound
default:
return nil
}
}
type grpcBatch struct {
database.BatchOps
db *GRPCClient
}
func (b *grpcBatch) Write() error {
request := &rpcdbpb.WriteBatchRequest{}
keySet := set.NewSet[string](len(b.Ops))
for i := len(b.Ops) - 1; i >= 0; i-- {
op := b.Ops[i]
key := string(op.Key)
if keySet.Contains(key) {
continue
}
keySet.Add(key)
if op.Delete {
request.Deletes = append(request.Deletes, &rpcdbpb.DeleteRequest{Key: op.Key})
} else {
request.Puts = append(request.Puts, &rpcdbpb.PutRequest{
Key: op.Key,
Value: op.Value,
})
}
}
resp, err := b.db.client.WriteBatch(context.Background(), request)
if err != nil {
return err
}
return codeToErr(resp.Err)
}
func (b *grpcBatch) Inner() database.Batch {
return b
}
type grpcIterator struct {
db *GRPCClient
id uint64
data []*rpcdbpb.PutRequest
fetchedData chan []*rpcdbpb.PutRequest
errLock sync.RWMutex
err error
reqUpdateError chan chan struct{}
once sync.Once
onClose chan struct{}
onClosed chan struct{}
}
func newGRPCIterator(db *GRPCClient, id uint64) *grpcIterator {
it := &grpcIterator{
db: db,
id: id,
fetchedData: make(chan []*rpcdbpb.PutRequest),
reqUpdateError: make(chan chan struct{}),
onClose: make(chan struct{}),
onClosed: make(chan struct{}),
}
go it.fetch()
return it
}
// Invariant: fetch is the only thread with access to send requests to
// the server's iterator. This is needed because iterators are not
// thread safe and the server expects the client (us) to only ever
// issue one request at a time for a given iterator id.
func (it *grpcIterator) fetch() {
defer func() {
resp, err := it.db.client.IteratorRelease(context.Background(), &rpcdbpb.IteratorReleaseRequest{Id: it.id})
if err != nil {
it.setError(err)
} else {
it.setError(codeToErr(resp.Err))
}
close(it.fetchedData)
close(it.onClosed)
}()
for {
resp, err := it.db.client.IteratorNext(context.Background(), &rpcdbpb.IteratorNextRequest{Id: it.id})
if err != nil {
it.setError(err)
return
}
if len(resp.Data) == 0 {
return
}
for {
select {
case it.fetchedData <- resp.Data:
case onUpdated := <-it.reqUpdateError:
it.updateError()
close(onUpdated)
continue
case <-it.onClose:
return
}
break
}
}
}
// Next attempts to move the iterator to the next element.
func (it *grpcIterator) Next() bool {
if it.db.closed.Get() {
it.data = nil
it.setError(database.ErrClosed)
return false
}
if len(it.data) > 1 {
it.data[0] = nil
it.data = it.data[1:]
return true
}
it.data = <-it.fetchedData
return len(it.data) > 0
}
// Error returns any error that occurred while iterating.
func (it *grpcIterator) Error() error {
if err := it.getError(); err != nil {
return err
}
onUpdated := make(chan struct{})
select {
case it.reqUpdateError <- onUpdated:
<-onUpdated
case <-it.onClosed:
}
return it.getError()
}
// Key returns the key of the current element.
func (it *grpcIterator) Key() []byte {
if len(it.data) == 0 {
return nil
}
return it.data[0].Key
}
// Value returns the value of the current element.
func (it *grpcIterator) Value() []byte {
if len(it.data) == 0 {
return nil
}
return it.data[0].Value
}
// Release frees any resources held by the iterator.
func (it *grpcIterator) Release() {
it.once.Do(func() {
close(it.onClose)
<-it.onClosed
})
}
func (it *grpcIterator) updateError() {
resp, err := it.db.client.IteratorError(context.Background(), &rpcdbpb.IteratorErrorRequest{Id: it.id})
if err != nil {
it.setError(err)
} else {
it.setError(codeToErr(resp.Err))
}
}
func (it *grpcIterator) setError(err error) {
if err == nil {
return
}
it.errLock.Lock()
defer it.errLock.Unlock()
if it.err == nil {
it.err = err
}
}
func (it *grpcIterator) getError() error {
it.errLock.RLock()
defer it.errLock.RUnlock()
return it.err
}
db/rpcdb/grpc_server.go
+2 -2
View File
@@ -12,13 +12,13 @@ import (
"github.com/luxfi/database"
rpcdbpb "github.com/luxfi/node/proto/pb/rpcdb"
rpcdb "github.com/luxfi/proto/rpcdb"
rpcdb "github.com/luxfi/protocol/rpcdb"
)
// GRPCServer is the gRPC transport adapter for the rpcdb Service. It
// wraps *Service and translates gRPC's protobuf-generated request /
// response types into the transport-neutral wire types in
// github.com/luxfi/proto/rpcdb. Pure adapter — no storage logic
// github.com/luxfi/protocol/rpcdb. Pure adapter — no storage logic
// lives here.
type GRPCServer struct {
rpcdbpb.UnimplementedDatabaseServer
internal/database/rpcdb/db_test.go → db/rpcdb/grpc_test.go
+25 -30
View File
@@ -1,6 +1,6 @@
//go:build grpc
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// Copyright (C) 2019-2026, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package rpcdb
@@ -21,15 +21,15 @@ import (
rpcdbpb "github.com/luxfi/node/proto/pb/rpcdb"
)
type testDatabase struct {
client *DatabaseClient
type testGRPCDatabase struct {
client *GRPCClient
server *memdb.Database
}
func setupDB(t testing.TB) *testDatabase {
func setupGRPCDB(t testing.TB) *testGRPCDatabase {
require := require.New(t)
db := &testDatabase{
db := &testGRPCDatabase{
server: memdb.New(),
}
@@ -38,7 +38,7 @@ func setupDB(t testing.TB) *testDatabase {
serverCloser := grpcutils.ServerCloser{}
server := grpcutils.NewServer()
rpcdbpb.RegisterDatabaseServer(server, NewServer(db.server))
rpcdbpb.RegisterDatabaseServer(server, NewGRPCServer(db.server))
serverCloser.Add(server)
go grpcutils.Serve(listener, server)
@@ -46,7 +46,7 @@ func setupDB(t testing.TB) *testDatabase {
conn, err := grpcutils.Dial(listener.Addr().String())
require.NoError(err)
db.client = NewClient(rpcdbpb.NewDatabaseClient(conn))
db.client = NewGRPCClient(rpcdbpb.NewDatabaseClient(conn))
t.Cleanup(func() {
serverCloser.Stop()
@@ -57,60 +57,57 @@ func setupDB(t testing.TB) *testDatabase {
return db
}
func TestInterface(t *testing.T) {
func TestGRPCInterface(t *testing.T) {
for name, test := range dbtest.Tests {
t.Run(name, func(t *testing.T) {
db := setupDB(t)
db := setupGRPCDB(t)
test(t, db.client)
})
}
}
func FuzzKeyValue(f *testing.F) {
db := setupDB(f)
func FuzzGRPCKeyValue(f *testing.F) {
db := setupGRPCDB(f)
dbtest.FuzzKeyValue(f, db.client)
}
func FuzzNewIteratorWithPrefix(f *testing.F) {
db := setupDB(f)
func FuzzGRPCNewIteratorWithPrefix(f *testing.F) {
db := setupGRPCDB(f)
dbtest.FuzzNewIteratorWithPrefix(f, db.client)
}
func FuzzNewIteratorWithStartAndPrefix(f *testing.F) {
db := setupDB(f)
func FuzzGRPCNewIteratorWithStartAndPrefix(f *testing.F) {
db := setupGRPCDB(f)
dbtest.FuzzNewIteratorWithStartAndPrefix(f, db.client)
}
func BenchmarkInterface(b *testing.B) {
func BenchmarkGRPCInterface(b *testing.B) {
for _, size := range dbtest.BenchmarkSizes {
keys, values := dbtest.SetupBenchmark(b, size[0], size[1], size[2])
for name, bench := range dbtest.Benchmarks {
b.Run(fmt.Sprintf("rpcdb_%d_pairs_%d_keys_%d_values_%s", size[0], size[1], size[2], name), func(b *testing.B) {
db := setupDB(b)
db := setupGRPCDB(b)
bench(b, db.client, keys, values)
})
}
}
}
func TestHealthCheck(t *testing.T) {
func TestGRPCHealthCheck(t *testing.T) {
scenarios := []struct {
name string
testDatabase *testDatabase
testFn func(db *corruptabledb.Database) error
wantErr bool
wantErrMsg string
name string
testFn func(db *corruptabledb.Database) error
wantErr bool
wantErrMsg string
}{
{
name: "healthcheck success",
testDatabase: setupDB(t),
name: "healthcheck success",
testFn: func(_ *corruptabledb.Database) error {
return nil
},
},
{
name: "healthcheck failed db closed",
testDatabase: setupDB(t),
name: "healthcheck failed db closed",
testFn: func(db *corruptabledb.Database) error {
return db.Close()
},
@@ -122,12 +119,11 @@ func TestHealthCheck(t *testing.T) {
t.Run(scenario.name, func(t *testing.T) {
require := require.New(t)
baseDB := setupDB(t)
baseDB := setupGRPCDB(t)
db := corruptabledb.New(baseDB.server, log.NoLog{})
defer db.Close()
require.NoError(scenario.testFn(db))
// check db HealthCheck
_, err := db.HealthCheck(context.Background())
if scenario.wantErr {
require.Error(err) //nolint:forbidigo
@@ -136,7 +132,6 @@ func TestHealthCheck(t *testing.T) {
}
require.NoError(err)
// check rpc HealthCheck
_, err = baseDB.client.HealthCheck(context.Background())
require.NoError(err)
})
db/rpcdb/service.go
+2 -2
View File
@@ -8,7 +8,7 @@
// Layered topology:
//
// Layer A — wire framing (github.com/luxfi/api/zap)
// Layer B — service spec (data carriers) (github.com/luxfi/proto/rpcdb)
// Layer B — service spec (data carriers) (github.com/luxfi/protocol/rpcdb)
// Layer C — service impl + transports (this package)
//
// One Service. Many transport adapters. Adding a new transport is a
@@ -27,7 +27,7 @@ import (
"sync"
"github.com/luxfi/database"
rpcdb "github.com/luxfi/proto/rpcdb"
rpcdb "github.com/luxfi/protocol/rpcdb"
)
// errUnknownIterator is the sentinel returned by IteratorNext / Error
db/rpcdb/zap_server.go
+1 -1
View File
@@ -12,7 +12,7 @@ import (
zapwire "github.com/luxfi/api/zap"
"github.com/luxfi/database"
rpcdb "github.com/luxfi/proto/rpcdb"
rpcdb "github.com/luxfi/protocol/rpcdb"
)
// ZAP db channel MsgType IDs. These are the wire-level Layer-A
go.mod
+1 -4
View File
@@ -196,7 +196,7 @@ require (
require (
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
github.com/luxfi/concurrent v0.0.3
github.com/luxfi/protocol v0.0.4 // indirect
github.com/luxfi/protocol v0.0.5
github.com/luxfi/upgrade v1.0.0 // indirect
github.com/luxfi/version v1.0.1
gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
@@ -226,7 +226,6 @@ require (
github.com/luxfi/formatting v1.0.1
github.com/luxfi/go-bip32 v1.0.2
github.com/luxfi/math/big v0.1.0 // indirect
github.com/luxfi/proto v0.0.0-00010101000000-000000000000
github.com/luxfi/sampler v1.0.0 // indirect
github.com/luxfi/tls v1.0.3 // indirect
github.com/mattn/go-colorable v0.1.14 // indirect
@@ -242,5 +241,3 @@ require (
)
exclude github.com/ethereum/go-ethereum v1.10.26
replace github.com/luxfi/proto => ../proto
go.sum
+4 -4
View File
@@ -262,8 +262,6 @@ github.com/luxfi/age v1.5.0 h1:G69HbSV4R3vKEH9B0CulnRaMdSdf4RalMgP8xKmxHeI=
github.com/luxfi/age v1.5.0/go.mod h1:iAYAxgvrXxcy746+Ovh/eWWDuF9teJLNcCSSOX9RYW0=
github.com/luxfi/ai v0.1.0 h1:PwTGob0GJivbdqNpUs82bvwcE8/qxyTokxfY7BZVPoo=
github.com/luxfi/ai v0.1.0/go.mod h1:lTuY32dGjEJUgVheYuanlt1O0jHj0AZ09sUSpRsmH6M=
github.com/luxfi/api v1.0.10 h1:tqtiCX8DcBsFG0JEhKy7eUNI+42+m2DoPjLjEh5TE5E=
github.com/luxfi/api v1.0.10/go.mod h1:5OFWvZF+PbyuvLCDyKKpXCA/xp+LxJ32yOhNEKwQFN0=
github.com/luxfi/api v1.0.11 h1:t4fzN9Ox/Vy5msW2WpSXq4xCAmBXXJS0oM+zIjM9IiQ=
github.com/luxfi/api v1.0.11/go.mod h1:q3Hh3mmkvp3cnv3aqe2+gCtmmSAL/uFjuzrqGavQLNA=
github.com/luxfi/atomic v1.0.0 h1:xUV60MuzRvXngaQ1sM0yVC2v4TRoLlUGkkH7M9PS4yw=
@@ -336,14 +334,16 @@ github.com/luxfi/pq v1.0.3 h1:pFlQm1+5FuKTDUh2y/23bXWkN4I2Rc5iuxJypwDFFMs=
github.com/luxfi/pq v1.0.3/go.mod h1:8bppZcRElfrVt0n3nYCZW3iX1TvhvzNbdjNdK1irgIE=
github.com/luxfi/precompile v0.5.17 h1:LlhU+eV+ZXhd8XWVKPLVPT6S3Xk0O8e3j60KurjnBuA=
github.com/luxfi/precompile v0.5.17/go.mod h1:zeuCoay3e1DI9rpNxMrJpE8+lKxaRGsV3tQ1N/O/a6Y=
github.com/luxfi/protocol v0.0.4 h1:wf3JSyeNMEabOUG0vExtIAtjfpJAyHXlepGBwv5g9NE=
github.com/luxfi/protocol v0.0.4/go.mod h1:9f35GLNlcHAz6LCvFBDZP5fTD2QnN0URXWGUOcD1JPU=
github.com/luxfi/protocol v0.0.5 h1:F3AoGtb9nEobmygnvLzVFYcU4F4wYWvG8f4HsHX4OdI=
github.com/luxfi/protocol v0.0.5/go.mod h1:7bKewL7MVngFF17nY4u/jwLLrXQOJf6pBwduI69t/9I=
github.com/luxfi/pulsar v1.0.0 h1:3R5QF9EzZ2IpwTD4AZmlEM0NIxenZEGHLFpBoQqBnIo=
github.com/luxfi/pulsar v1.0.0/go.mod h1:6bT5BfrAS3lI3qPbcRHEclCgMZmo0XjhjCTFvGuboIk=
github.com/luxfi/relay v0.0.0-20260429020048-c629fe160d3c h1:wiOLqdEEjIoq+Uubkm6F6GYQue+rzqnITScM3ohWGqI=
github.com/luxfi/relay v0.0.0-20260429020048-c629fe160d3c/go.mod h1:P408Sn9NGwtcDOoGdkmWe0484dkWDvEvDDBM44Cv+s4=
github.com/luxfi/resource v0.0.1 h1:mTh+ICWSy548GTUSSyx7V/X5dV18oEwxZeQEYGJQhD4=
github.com/luxfi/resource v0.0.1/go.mod h1:wWpZktciYwIi6RNqA+fHwzmPrUJa7PRX7urfwT+spRE=
github.com/luxfi/corona v0.2.0 h1:DbLMZmE//2T2wXXS/gEkKZrTbre9LD+7EE/tz5SQszU=
github.com/luxfi/corona v0.2.0/go.mod h1:SZ+aDLUdfSAtaTRaaYzeDZ5DNQiLaOCelSaIGjL21R0=
github.com/luxfi/rpc v1.0.2 h1:NLRcOYRW+io0d1d33RMkgOZea8nlhK09MbPgCXcU5wU=
github.com/luxfi/rpc v1.0.2/go.mod h1:pgiHwMWgOuxYYIa0vsUBvrBI+Op6bhZ39guM9vtMUcE=
github.com/luxfi/runtime v1.0.1 h1:cii3OsRiVSIl9jzfWFM6++62T1r6ZSGP+/3F0Ezhpqw=
internal/database/rpcdb/db_client.go
-373
View File
@@ -1,373 +0,0 @@
//go:build grpc
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package rpcdb
import (
"context"
"encoding/json"
"errors"
"io"
"sync"
"google.golang.org/protobuf/types/known/emptypb"
"github.com/luxfi/database"
"github.com/luxfi/math/set"
"github.com/luxfi/utils"
rpcdbpb "github.com/luxfi/node/proto/pb/rpcdb"
)
var (
_ database.Database = (*DatabaseClient)(nil)
_ database.Batch = (*batch)(nil)
_ database.Iterator = (*iterator)(nil)
)
// DatabaseClient is an implementation of database that talks over RPC.
type DatabaseClient struct {
client rpcdbpb.DatabaseClient
closed utils.Atomic[bool]
}
// NewClient returns a database instance connected to a remote database instance
func NewClient(client rpcdbpb.DatabaseClient) *DatabaseClient {
return &DatabaseClient{client: client}
}
// Has attempts to return if the database has a key with the provided value.
func (db *DatabaseClient) Has(key []byte) (bool, error) {
resp, err := db.client.Has(context.Background(), &rpcdbpb.HasRequest{
Key: key,
})
if err != nil {
return false, err
}
return resp.Has, ErrEnumToError[resp.Err]
}
// Get attempts to return the value that was mapped to the key that was provided
func (db *DatabaseClient) Get(key []byte) ([]byte, error) {
resp, err := db.client.Get(context.Background(), &rpcdbpb.GetRequest{
Key: key,
})
if err != nil {
return nil, err
}
return resp.Value, ErrEnumToError[resp.Err]
}
// Put attempts to set the value this key maps to
func (db *DatabaseClient) Put(key, value []byte) error {
resp, err := db.client.Put(context.Background(), &rpcdbpb.PutRequest{
Key: key,
Value: value,
})
if err != nil {
return err
}
return ErrEnumToError[resp.Err]
}
// Delete attempts to remove any mapping from the key
func (db *DatabaseClient) Delete(key []byte) error {
resp, err := db.client.Delete(context.Background(), &rpcdbpb.DeleteRequest{
Key: key,
})
if err != nil {
return err
}
return ErrEnumToError[resp.Err]
}
// NewBatch returns a new batch
func (db *DatabaseClient) NewBatch() database.Batch {
return &batch{db: db}
}
func (db *DatabaseClient) NewIterator() database.Iterator {
return db.NewIteratorWithStartAndPrefix(nil, nil)
}
func (db *DatabaseClient) NewIteratorWithStart(start []byte) database.Iterator {
return db.NewIteratorWithStartAndPrefix(start, nil)
}
func (db *DatabaseClient) NewIteratorWithPrefix(prefix []byte) database.Iterator {
return db.NewIteratorWithStartAndPrefix(nil, prefix)
}
// NewIteratorWithStartAndPrefix returns a new empty iterator
func (db *DatabaseClient) NewIteratorWithStartAndPrefix(start, prefix []byte) database.Iterator {
resp, err := db.client.NewIteratorWithStartAndPrefix(context.Background(), &rpcdbpb.NewIteratorWithStartAndPrefixRequest{
Start: start,
Prefix: prefix,
})
if err != nil {
return &database.IteratorError{
Err: err,
}
}
return newIterator(db, resp.Id)
}
// Compact attempts to optimize the space utilization in the provided range
func (db *DatabaseClient) Compact(start, limit []byte) error {
resp, err := db.client.Compact(context.Background(), &rpcdbpb.CompactRequest{
Start: start,
Limit: limit,
})
if err != nil {
return err
}
return ErrEnumToError[resp.Err]
}
// Close attempts to close the database
func (db *DatabaseClient) Close() error {
db.closed.Set(true)
resp, err := db.client.Close(context.Background(), &rpcdbpb.CloseRequest{})
if err != nil {
return err
}
return ErrEnumToError[resp.Err]
}
// Sync flushes any pending writes to persistent storage.
// For RPC databases, this delegates to the remote database's Sync.
func (db *DatabaseClient) Sync() error {
// RPC database delegates sync to the underlying database on the server side.
// Most operations are already synchronous over RPC, so this is typically a no-op.
// If the server implements a Sync RPC method, call it here.
// For now, return nil as operations are synchronous.
return nil
}
func (db *DatabaseClient) HealthCheck(ctx context.Context) (interface{}, error) {
health, err := db.client.HealthCheck(ctx, &emptypb.Empty{})
if err != nil {
return nil, err
}
return json.RawMessage(health.Details), nil
}
// Backup is not supported over the RPC database client.
func (db *DatabaseClient) Backup(_ io.Writer, _ uint64) (uint64, error) {
return 0, errors.New("rpcdb: backup not supported")
}
// Load is not supported over the RPC database client.
func (db *DatabaseClient) Load(_ io.Reader) error {
return errors.New("rpcdb: load not supported")
}
type batch struct {
database.BatchOps
db *DatabaseClient
}
func (b *batch) Write() error {
request := &rpcdbpb.WriteBatchRequest{}
keySet := set.NewSet[string](len(b.Ops))
for i := len(b.Ops) - 1; i >= 0; i-- {
op := b.Ops[i]
key := string(op.Key)
if keySet.Contains(key) {
continue
}
keySet.Add(key)
if op.Delete {
request.Deletes = append(request.Deletes, &rpcdbpb.DeleteRequest{
Key: op.Key,
})
} else {
request.Puts = append(request.Puts, &rpcdbpb.PutRequest{
Key: op.Key,
Value: op.Value,
})
}
}
resp, err := b.db.client.WriteBatch(context.Background(), request)
if err != nil {
return err
}
return ErrEnumToError[resp.Err]
}
func (b *batch) Inner() database.Batch {
return b
}
type iterator struct {
db *DatabaseClient
id uint64
data []*rpcdbpb.PutRequest
fetchedData chan []*rpcdbpb.PutRequest
errLock sync.RWMutex
err error
reqUpdateError chan chan struct{}
once sync.Once
onClose chan struct{}
onClosed chan struct{}
}
func newIterator(db *DatabaseClient, id uint64) *iterator {
it := &iterator{
db: db,
id: id,
fetchedData: make(chan []*rpcdbpb.PutRequest),
reqUpdateError: make(chan chan struct{}),
onClose: make(chan struct{}),
onClosed: make(chan struct{}),
}
go it.fetch()
return it
}
// Invariant: fetch is the only thread with access to send requests to the
// server's iterator. This is needed because iterators are not thread safe and
// the server expects the client (us) to only ever issue one request at a time
// for a given iterator id.
func (it *iterator) fetch() {
defer func() {
resp, err := it.db.client.IteratorRelease(context.Background(), &rpcdbpb.IteratorReleaseRequest{
Id: it.id,
})
if err != nil {
it.setError(err)
} else {
it.setError(ErrEnumToError[resp.Err])
}
close(it.fetchedData)
close(it.onClosed)
}()
for {
resp, err := it.db.client.IteratorNext(context.Background(), &rpcdbpb.IteratorNextRequest{
Id: it.id,
})
if err != nil {
it.setError(err)
return
}
if len(resp.Data) == 0 {
return
}
for {
select {
case it.fetchedData <- resp.Data:
case onUpdated := <-it.reqUpdateError:
it.updateError()
close(onUpdated)
continue
case <-it.onClose:
return
}
break
}
}
}
// Next attempts to move the iterator to the next element and returns if this
// succeeded
func (it *iterator) Next() bool {
if it.db.closed.Get() {
it.data = nil
it.setError(database.ErrClosed)
return false
}
if len(it.data) > 1 {
it.data[0] = nil
it.data = it.data[1:]
return true
}
it.data = <-it.fetchedData
return len(it.data) > 0
}
// Error returns any that occurred while iterating
func (it *iterator) Error() error {
if err := it.getError(); err != nil {
return err
}
onUpdated := make(chan struct{})
select {
case it.reqUpdateError <- onUpdated:
<-onUpdated
case <-it.onClosed:
}
return it.getError()
}
// Key returns the key of the current element
func (it *iterator) Key() []byte {
if len(it.data) == 0 {
return nil
}
return it.data[0].Key
}
// Value returns the value of the current element
func (it *iterator) Value() []byte {
if len(it.data) == 0 {
return nil
}
return it.data[0].Value
}
// Release frees any resources held by the iterator
func (it *iterator) Release() {
it.once.Do(func() {
close(it.onClose)
<-it.onClosed
})
}
func (it *iterator) updateError() {
resp, err := it.db.client.IteratorError(context.Background(), &rpcdbpb.IteratorErrorRequest{
Id: it.id,
})
if err != nil {
it.setError(err)
} else {
it.setError(ErrEnumToError[resp.Err])
}
}
func (it *iterator) setError(err error) {
if err == nil {
return
}
it.errLock.Lock()
defer it.errLock.Unlock()
if it.err == nil {
it.err = err
}
}
func (it *iterator) getError() error {
it.errLock.RLock()
defer it.errLock.RUnlock()
return it.err
}
internal/database/rpcdb/db_server.go
-199
View File
@@ -1,199 +0,0 @@
//go:build grpc
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package rpcdb
import (
"context"
"encoding/json"
"errors"
"sync"
"google.golang.org/protobuf/types/known/emptypb"
"github.com/luxfi/constants"
"github.com/luxfi/database"
rpcdbpb "github.com/luxfi/node/proto/pb/rpcdb"
)
const iterationBatchSize = 128 * constants.KiB
var errUnknownIterator = errors.New("unknown iterator")
// DatabaseServer is a database that is managed over RPC.
type DatabaseServer struct {
rpcdbpb.UnsafeDatabaseServer
db database.Database
// iteratorLock protects [nextIteratorID] and [iterators] from concurrent
// modifications. Similarly to [batchLock], [iteratorLock] does not protect
// the actual Iterator. Iterators are documented as not being safe for
// concurrent use. Therefore, it is up to the client to respect this
// invariant.
iteratorLock sync.RWMutex
nextIteratorID uint64
iterators map[uint64]database.Iterator
}
// NewServer returns a database instance that is managed remotely
func NewServer(db database.Database) *DatabaseServer {
return &DatabaseServer{
db: db,
iterators: make(map[uint64]database.Iterator),
}
}
// Has delegates the Has call to the managed database and returns the result
func (db *DatabaseServer) Has(_ context.Context, req *rpcdbpb.HasRequest) (*rpcdbpb.HasResponse, error) {
has, err := db.db.Has(req.Key)
return &rpcdbpb.HasResponse{
Has: has,
Err: ErrorToErrEnum[err],
}, ErrorToRPCError(err)
}
// Get delegates the Get call to the managed database and returns the result
func (db *DatabaseServer) Get(_ context.Context, req *rpcdbpb.GetRequest) (*rpcdbpb.GetResponse, error) {
value, err := db.db.Get(req.Key)
return &rpcdbpb.GetResponse{
Value: value,
Err: ErrorToErrEnum[err],
}, ErrorToRPCError(err)
}
// Put delegates the Put call to the managed database and returns the result
func (db *DatabaseServer) Put(_ context.Context, req *rpcdbpb.PutRequest) (*rpcdbpb.PutResponse, error) {
err := db.db.Put(req.Key, req.Value)
return &rpcdbpb.PutResponse{Err: ErrorToErrEnum[err]}, ErrorToRPCError(err)
}
// Delete delegates the Delete call to the managed database and returns the
// result
func (db *DatabaseServer) Delete(_ context.Context, req *rpcdbpb.DeleteRequest) (*rpcdbpb.DeleteResponse, error) {
err := db.db.Delete(req.Key)
return &rpcdbpb.DeleteResponse{Err: ErrorToErrEnum[err]}, ErrorToRPCError(err)
}
// Compact delegates the Compact call to the managed database and returns the
// result
func (db *DatabaseServer) Compact(_ context.Context, req *rpcdbpb.CompactRequest) (*rpcdbpb.CompactResponse, error) {
err := db.db.Compact(req.Start, req.Limit)
return &rpcdbpb.CompactResponse{Err: ErrorToErrEnum[err]}, ErrorToRPCError(err)
}
// Close delegates the Close call to the managed database and returns the result
func (db *DatabaseServer) Close(context.Context, *rpcdbpb.CloseRequest) (*rpcdbpb.CloseResponse, error) {
err := db.db.Close()
return &rpcdbpb.CloseResponse{Err: ErrorToErrEnum[err]}, ErrorToRPCError(err)
}
// HealthCheck performs a heath check against the underlying database.
func (db *DatabaseServer) HealthCheck(ctx context.Context, _ *emptypb.Empty) (*rpcdbpb.HealthCheckResponse, error) {
health, err := db.db.HealthCheck(ctx)
if err != nil {
return &rpcdbpb.HealthCheckResponse{}, err
}
details, err := json.Marshal(health)
return &rpcdbpb.HealthCheckResponse{
Details: details,
}, err
}
// WriteBatch takes in a set of key-value pairs and atomically writes them to
// the internal database
func (db *DatabaseServer) WriteBatch(_ context.Context, req *rpcdbpb.WriteBatchRequest) (*rpcdbpb.WriteBatchResponse, error) {
batch := db.db.NewBatch()
for _, put := range req.Puts {
if err := batch.Put(put.Key, put.Value); err != nil {
return &rpcdbpb.WriteBatchResponse{
Err: ErrorToErrEnum[err],
}, ErrorToRPCError(err)
}
}
for _, del := range req.Deletes {
if err := batch.Delete(del.Key); err != nil {
return &rpcdbpb.WriteBatchResponse{
Err: ErrorToErrEnum[err],
}, ErrorToRPCError(err)
}
}
err := batch.Write()
return &rpcdbpb.WriteBatchResponse{
Err: ErrorToErrEnum[err],
}, ErrorToRPCError(err)
}
// NewIteratorWithStartAndPrefix allocates an iterator and returns the iterator
// ID
func (db *DatabaseServer) NewIteratorWithStartAndPrefix(_ context.Context, req *rpcdbpb.NewIteratorWithStartAndPrefixRequest) (*rpcdbpb.NewIteratorWithStartAndPrefixResponse, error) {
it := db.db.NewIteratorWithStartAndPrefix(req.Start, req.Prefix)
db.iteratorLock.Lock()
defer db.iteratorLock.Unlock()
id := db.nextIteratorID
db.iterators[id] = it
db.nextIteratorID++
return &rpcdbpb.NewIteratorWithStartAndPrefixResponse{Id: id}, nil
}
// IteratorNext attempts to call next on the requested iterator
func (db *DatabaseServer) IteratorNext(_ context.Context, req *rpcdbpb.IteratorNextRequest) (*rpcdbpb.IteratorNextResponse, error) {
db.iteratorLock.RLock()
it, exists := db.iterators[req.Id]
db.iteratorLock.RUnlock()
if !exists {
return nil, errUnknownIterator
}
var (
size int
data []*rpcdbpb.PutRequest
)
for size < iterationBatchSize && it.Next() {
key := it.Key()
value := it.Value()
size += len(key) + len(value)
data = append(data, &rpcdbpb.PutRequest{
Key: key,
Value: value,
})
}
return &rpcdbpb.IteratorNextResponse{Data: data}, nil
}
// IteratorError attempts to report any errors that occurred during iteration
func (db *DatabaseServer) IteratorError(_ context.Context, req *rpcdbpb.IteratorErrorRequest) (*rpcdbpb.IteratorErrorResponse, error) {
db.iteratorLock.RLock()
it, exists := db.iterators[req.Id]
db.iteratorLock.RUnlock()
if !exists {
return nil, errUnknownIterator
}
err := it.Error()
return &rpcdbpb.IteratorErrorResponse{Err: ErrorToErrEnum[err]}, ErrorToRPCError(err)
}
// IteratorRelease attempts to release the resources allocated to an iterator
func (db *DatabaseServer) IteratorRelease(_ context.Context, req *rpcdbpb.IteratorReleaseRequest) (*rpcdbpb.IteratorReleaseResponse, error) {
db.iteratorLock.Lock()
it, exists := db.iterators[req.Id]
if !exists {
db.iteratorLock.Unlock()
return &rpcdbpb.IteratorReleaseResponse{}, nil
}
delete(db.iterators, req.Id)
db.iteratorLock.Unlock()
err := it.Error()
it.Release()
return &rpcdbpb.IteratorReleaseResponse{Err: ErrorToErrEnum[err]}, ErrorToRPCError(err)
}
internal/database/rpcdb/errors.go
-30
View File
@@ -1,30 +0,0 @@
//go:build grpc
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package rpcdb
import (
"github.com/luxfi/database"
rpcdbpb "github.com/luxfi/node/proto/pb/rpcdb"
)
var (
ErrEnumToError = map[rpcdbpb.Error]error{
rpcdbpb.Error_ERROR_CLOSED: database.ErrClosed,
rpcdbpb.Error_ERROR_NOT_FOUND: database.ErrNotFound,
}
ErrorToErrEnum = map[error]rpcdbpb.Error{
database.ErrClosed: rpcdbpb.Error_ERROR_CLOSED,
database.ErrNotFound: rpcdbpb.Error_ERROR_NOT_FOUND,
}
)
func ErrorToRPCError(err error) error {
if _, ok := ErrorToErrEnum[err]; ok {
return nil
}
return err
}
service/keystore/rpckeystore/client.go
+2 -2
View File
@@ -16,8 +16,8 @@ import (
"github.com/luxfi/database"
"github.com/luxfi/database/encdb"
"github.com/luxfi/node/db/rpcdb"
"github.com/luxfi/node/service/keystore"
"github.com/luxfi/node/internal/database/rpcdb"
"github.com/luxfi/vm/rpc/grpcutils"
keystorepb "github.com/luxfi/node/proto/pb/keystore"
@@ -60,6 +60,6 @@ func (c *Client) GetRawDatabase(username, password string) (database.Database, e
return nil, err
}
dbClient := rpcdb.NewClient(rpcdbpb.NewDatabaseClient(clientConn))
dbClient := rpcdb.NewGRPCClient(rpcdbpb.NewDatabaseClient(clientConn))
return dbClient, nil
}