merge: branch '3411-serve-db' into 'main'

Database debug changes [#3411]

Closes #3411

See merge request accumulatenetwork/accumulate!919

.gitignore

@@ -13,7 +13,7 @@ tools/cmd/gen-enum/go_build*
 *.exe
 *.test
 .testdata
-__debug_bin
+__debug_bin*
 tmp
 
 # Temp and log files

.vscode/launch.json

@@ -65,6 +65,27 @@
         "bad",
         "diff.dat"
       ],
+    },
+    // Build Missing file
+    {
+      "name": "Build Missing file",
+      "presentation": {
+        "group": "00_DB_Repair"
+      },
+      "type": "go",
+      "request": "launch",
+      "mode": "auto",
+      "program": "${workspaceFolder}/tools/cmd/dbrepair",
+      "dlvFlags": [
+        "--check-go-version=false"
+      ],
+      "cwd": "${workspaceFolder}",
+      "args": [
+        "buildMissing",
+        "missing.dat",
+        "good",
+        "fix.dat"
+      ],
     },
       // Build Fix file
       {

cmd/accumulated-bootstrap/main.go

@@ -18,7 +18,7 @@ import (
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	"github.com/multiformats/go-multiaddr"
 	"github.com/spf13/cobra"
-	. "gitlab.com/accumulatenetwork/accumulate/cmd/internal"
+	. "gitlab.com/accumulatenetwork/accumulate/internal/util/cmd"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3/p2p"
 )
 

cmd/accumulated-faucet/main.go

@@ -19,10 +19,10 @@ import (
 	"github.com/multiformats/go-multiaddr"
 	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
-	. "gitlab.com/accumulatenetwork/accumulate/cmd/internal"
 	"gitlab.com/accumulatenetwork/accumulate/internal/api/routing"
 	v3impl "gitlab.com/accumulatenetwork/accumulate/internal/api/v3"
 	"gitlab.com/accumulatenetwork/accumulate/internal/logging"
+	. "gitlab.com/accumulatenetwork/accumulate/internal/util/cmd"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3/message"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3/p2p"

cmd/accumulated-http/main.go

@@ -24,13 +24,11 @@ import (
 	"github.com/cometbft/cometbft/libs/log"
 	"github.com/multiformats/go-multiaddr"
 	"github.com/rs/cors"
-	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
-	. "gitlab.com/accumulatenetwork/accumulate/cmd/internal"
 	"gitlab.com/accumulatenetwork/accumulate/internal/api/routing"
-	"gitlab.com/accumulatenetwork/accumulate/internal/logging"
 	accumulated "gitlab.com/accumulatenetwork/accumulate/internal/node/daemon"
 	nodehttp "gitlab.com/accumulatenetwork/accumulate/internal/node/http"
+	. "gitlab.com/accumulatenetwork/accumulate/internal/util/cmd"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3/message"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3/p2p"
@@ -102,12 +100,7 @@ func run(_ *cobra.Command, args []string) {
 		Fatalf("must specify at least one peer")
 	}
 
-	lw, err := logging.NewConsoleWriter("plain")
-	Check(err)
-	ll, lw, err := logging.ParseLogLevel(flag.LogLevel, lw)
-	Check(err)
-	logger, err := logging.NewTendermintLogger(zerolog.New(lw), ll, false)
-	Check(err)
+	logger := NewConsoleLogger(flag.LogLevel)
 
 	node, err := p2p.New(p2p.Options{
 		Key:               loadOrGenerateKey(),

exp/light/sync.go

@@ -648,6 +648,7 @@ func (c *Client) IndexAccountTransactions(ctx context.Context, accounts ...*url.
 	// Load the root chain and anchor index of each partition, and the
 	// directory's anchor chain index for each partition
 	type PartData struct {
+		ID                        string
 		RootIndex, DirAnchorIndex []*protocol.IndexEntry
 		AnchorIndex               []*AnchorMetadata
 	}
@@ -667,6 +668,7 @@ func (c *Client) IndexAccountTransactions(ctx context.Context, accounts ...*url.
 		}
 
 		part := new(PartData)
+		part.ID = partId
 		parts[partId] = part
 
 		// Load the partition's root chain index
@@ -754,7 +756,7 @@ func (c *Client) IndexAccountTransactions(ctx context.Context, accounts ...*url.
 				// Get anchor
 				_, anchor, ok := FindEntry(part.AnchorIndex, ByAnchorBlock(root.BlockIndex))
 				if !ok || anchor.Anchor.MinorBlockIndex != root.BlockIndex {
-					return errors.NotFound.WithFormat("cannot find anchor for block %d", root.BlockIndex)
+					return errors.NotFound.WithFormat("cannot find anchor for %s block %d", part.ID, root.BlockIndex)
 				}
 
 				// TODO skip if not anchored

internal/api/v2/api.go

@@ -41,7 +41,6 @@ type Options struct {
 	LocalV3       V3
 	Querier       api.Querier
 	Submitter     api.Submitter
-	Network       api.NetworkService
 	Faucet        api.Faucet
 	Validator     api.Validator
 	Sequencer     private.Sequencer

internal/api/v2/jrpc.go

@@ -9,6 +9,7 @@ package api
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	stdlog "log"
 	"mime"
@@ -20,7 +21,6 @@ import (
 	"github.com/go-playground/validator/v10"
 	"gitlab.com/accumulatenetwork/accumulate"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/api/v3"
-	"gitlab.com/accumulatenetwork/accumulate/pkg/errors"
 	"gitlab.com/accumulatenetwork/accumulate/protocol"
 )
 
@@ -40,16 +40,6 @@ func NewJrpc(opts Options) (*JrpcMethods, error) {
 		m.logger = opts.Logger.With("module", "jrpc")
 	}
 
-	if opts.LocalV3 == nil ||
-		opts.Querier == nil ||
-		opts.Submitter == nil ||
-		opts.Network == nil ||
-		opts.Faucet == nil ||
-		opts.Validator == nil ||
-		opts.Sequencer == nil {
-		return nil, errors.BadRequest.With("missing P2P clients")
-	}
-
 	m.validate, err = protocol.NewValidator()
 	if err != nil {
 		return nil, err
@@ -102,6 +92,10 @@ func (m *JrpcMethods) jrpc2http(jrpc jsonrpc2.MethodFunc) http.HandlerFunc {
 }
 
 func (m *JrpcMethods) Status(ctx context.Context, _ json.RawMessage) interface{} {
+	if m.LocalV3 == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	s, err := m.LocalV3.ConsensusStatus(ctx, api.ConsensusStatusOptions{})
 	if err != nil {
 		return accumulateError(err)
@@ -137,6 +131,10 @@ func (m *JrpcMethods) Version(ctx context.Context, _ json.RawMessage) interface{
 }
 
 func (m *JrpcMethods) Describe(ctx context.Context, _ json.RawMessage) interface{} {
+	if m.LocalV3 == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	net, err := m.LocalV3.NetworkStatus(ctx, api.NetworkStatusOptions{})
 	if err != nil {
 		return accumulateError(err)

internal/api/v2/jrpc_execute.go

@@ -69,6 +69,10 @@ func (m *JrpcMethods) executeWith(ctx context.Context, params json.RawMessage, p
 }
 
 func (m *JrpcMethods) Faucet(ctx context.Context, params json.RawMessage) interface{} {
+	if m.Options.Faucet == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(protocol.AcmeFaucet)
 	err := m.parse(params, req)
 	if err != nil {
@@ -161,6 +165,10 @@ func (m *JrpcMethods) ExecuteDirect(ctx context.Context, params json.RawMessage)
 }
 
 func (m *JrpcMethods) ExecuteLocal(ctx context.Context, params json.RawMessage) interface{} {
+	if m.LocalV3 == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(ExecuteRequest)
 	err := json.Unmarshal(params, req)
 	if err != nil {
@@ -181,8 +189,14 @@ func (m *JrpcMethods) submit(sub api.Submitter, val api.Validator, ctx context.C
 	var resp []*api.Submission
 	var yes, no = true, false
 	if checkOnly {
+		if val == nil {
+			return accumulateError(fmt.Errorf("service not available"))
+		}
 		resp, err = val.Validate(ctx, env, api.ValidateOptions{Full: &no})
 	} else {
+		if sub == nil {
+			return accumulateError(fmt.Errorf("service not available"))
+		}
 		resp, err = sub.Submit(ctx, env, api.SubmitOptions{Verify: &no, Wait: &yes})
 	}
 	if err != nil {

internal/api/v2/jrpc_metrics.go

@@ -18,6 +18,10 @@ import (
 
 // Metrics returns Metrics for explorer (tps, etc.)
 func (m *JrpcMethods) Metrics(ctx context.Context, params json.RawMessage) interface{} {
+	if m.LocalV3 == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(MetricsQuery)
 	err := m.parse(params, req)
 	if err != nil {

internal/api/v2/query_jrpc.go

@@ -9,6 +9,7 @@ package api
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"sync"
 	"sync/atomic"
 
@@ -18,6 +19,10 @@ import (
 )
 
 func (m *JrpcMethods) QueryTx(ctx context.Context, params json.RawMessage) interface{} {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(TxnQuery)
 	err := m.parse(params, req)
 	if err != nil {

internal/api/v2/query_v3.go

@@ -311,6 +311,10 @@ func queryTx(v3 api.Querier, ctx context.Context, txid *url.TxID, includeReceipt
 }
 
 func (m *JrpcMethods) QueryDirectory(ctx context.Context, params json.RawMessage) any {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(DirectoryQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -602,6 +606,10 @@ func txnOrSigV3(v3 api.Querier, ctx context.Context, r *api.ChainEntryRecord[*ap
 }
 
 func (m *JrpcMethods) QueryKeyPageIndex(ctx context.Context, params json.RawMessage) any {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(KeyPageIndexQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -637,6 +645,10 @@ func (m *JrpcMethods) QueryKeyPageIndex(ctx context.Context, params json.RawMess
 }
 
 func (m *JrpcMethods) QueryData(ctx context.Context, params json.RawMessage) any {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(DataEntryQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -668,6 +680,10 @@ func (m *JrpcMethods) QueryData(ctx context.Context, params json.RawMessage) any
 }
 
 func (m *JrpcMethods) QueryDataSet(ctx context.Context, params json.RawMessage) any {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(DataEntrySetQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -691,6 +707,10 @@ func (m *JrpcMethods) QueryDataSet(ctx context.Context, params json.RawMessage)
 }
 
 func (m *JrpcMethods) QueryTxHistory(ctx context.Context, params json.RawMessage) any {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(TxHistoryQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -730,6 +750,10 @@ func (m *JrpcMethods) QueryTxHistory(ctx context.Context, params json.RawMessage
 }
 
 func (m *JrpcMethods) QueryTxLocal(ctx context.Context, params json.RawMessage) any {
+	if m.LocalV3 == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(TxnQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -755,6 +779,10 @@ func (m *JrpcMethods) QueryTxLocal(ctx context.Context, params json.RawMessage)
 
 // Query queries an account or account chain by URL.
 func (m *JrpcMethods) Query(ctx context.Context, params json.RawMessage) any {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(GeneralQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -1013,6 +1041,10 @@ chain_query:
 }
 
 func (m *JrpcMethods) QueryMinorBlocks(ctx context.Context, params json.RawMessage) interface{} {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(MinorBlocksQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -1114,6 +1146,10 @@ func (m *JrpcMethods) QueryMinorBlocks(ctx context.Context, params json.RawMessa
 }
 
 func (m *JrpcMethods) QueryMajorBlocks(ctx context.Context, params json.RawMessage) interface{} {
+	if m.Querier == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(MajorBlocksQuery)
 	err := m.parse(params, req)
 	if err != nil {
@@ -1159,6 +1195,10 @@ func (m *JrpcMethods) QueryMajorBlocks(ctx context.Context, params json.RawMessa
 }
 
 func (m *JrpcMethods) QuerySynth(ctx context.Context, params json.RawMessage) interface{} {
+	if m.Sequencer == nil {
+		return accumulateError(fmt.Errorf("service not available"))
+	}
+
 	req := new(SyntheticTransactionRequest)
 	err := m.parse(params, req)
 	if err != nil {

internal/core/execute/v1/simulator/simulator.go

@@ -613,7 +613,6 @@ func (x *ExecEntry) init(sim *Simulator, logger log.Logger, partition *config.Pa
 		LocalV3:       x.service,
 		Querier:       sim.Services(),
 		Submitter:     sim.Services(),
-		Network:       sim.Services(),
 		Faucet:        sim.Services(),
 		Validator:     sim.Services(),
 		Sequencer:     sim.Services(),

internal/database/account_chains.go

@@ -44,7 +44,7 @@ func newChain2(parent record.Record, _ log.Logger, _ record.Store, key *record.K
 		"SignatureChain",
 		"ScratchChain",
 		"AnchorSequenceChain",
-		"SyntheticSequenceChain":
+		"SyntheticSequenceChain": // Bug, this is actually an index chain
 		typ = merkle.ChainTypeTransaction
 	case "RootChain",
 		"AnchorChain":

internal/node/http/handler.go

@@ -113,7 +113,6 @@ func NewHandler(opts Options) (*Handler, error) {
 		LocalV3:       selfClient,
 		Querier:       &api.Collator{Querier: client, Network: client},
 		Submitter:     client,
-		Network:       client,
 		Faucet:        client,
 		Validator:     client,
 		Sequencer:     client.Private(),

cmd/internal/errors.go → internal/util/cmd/errors.go

@@ -4,7 +4,7 @@
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
-package internal
+package cmdutil
 
 import (
 	"fmt"

cmd/internal/flags.go → internal/util/cmd/flags.go

@@ -4,7 +4,7 @@
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
-package internal
+package cmdutil
 
 import (
 	"fmt"