feat: re-broadcast stale txs

Makefile

@@ -25,18 +25,18 @@ build_docker:
 
 .PHONY: run
 run:
-	docker compose down --remove-orphans
+	docker compose down -v --remove-orphans
 	docker compose up --build blocktx callbacker metamorph api
 
 .PHONY: run_e2e_tests
 run_e2e_tests:
-	docker compose down --remove-orphans
+	docker compose down -v --remove-orphans
 	docker compose up --build blocktx callbacker metamorph api tests --scale blocktx=4 --scale metamorph=2 --exit-code-from tests
 	docker compose down
 
 .PHONY: run_e2e_tests_with_tracing
 run_e2e_tests_with_tracing:
-	docker compose down --remove-orphans
+	docker compose down -v --remove-orphans
 	ARC_TRACING_ENABLED=TRUE docker compose up --build blocktx callbacker metamorph api tests jaeger --scale blocktx=4 --scale metamorph=2 --no-attach jaeger
 
 .PHONY: test

cmd/arc/services/blocktx.go

@@ -8,12 +8,13 @@ import (
 
 	"go.opentelemetry.io/otel/attribute"
 
+	"github.com/libsv/go-p2p"
+
 	"github.com/bitcoin-sv/arc/internal/grpc_opts"
 	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_core"
 	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_jetstream"
 	"github.com/bitcoin-sv/arc/internal/message_queue/nats/nats_connection"
 	"github.com/bitcoin-sv/arc/internal/tracing"
-	"github.com/libsv/go-p2p"
 
 	"github.com/bitcoin-sv/arc/config"
 	"github.com/bitcoin-sv/arc/internal/blocktx"
@@ -127,6 +128,7 @@ func StartBlockTx(logger *slog.Logger, arcConfig *config.ArcConfig) (func(), err
 		blocktx.WithRegisterTxsInterval(btxConfig.RegisterTxsInterval),
 		blocktx.WithMessageQueueClient(mqClient),
 		blocktx.WithMaxBlockProcessingDuration(btxConfig.MaxBlockProcessingDuration),
+		blocktx.WithIncomingIsLongest(btxConfig.IncomingIsLongest),
 	)
 
 	blockRequestCh := make(chan blocktx.BlockRequest, blockProcessingBuffer)

config/config.go

@@ -128,6 +128,7 @@ type BlocktxConfig struct {
 	MaxAllowedBlockHeightMismatch int                 `mapstructure:"maxAllowedBlockHeightMismatch"`
 	MessageQueue                  *MessageQueueConfig `mapstructure:"mq"`
 	P2pReadBufferSize             int                 `mapstructure:"p2pReadBufferSize"`
+	IncomingIsLongest             bool                `mapstructure:"incomingIsLongest"`
 }
 
 type DbConfig struct {

config/defaults.go

@@ -129,6 +129,7 @@ func getBlocktxConfig() *BlocktxConfig {
 		MaxBlockProcessingDuration:    5 * time.Minute,
 		MessageQueue:                  &MessageQueueConfig{},
 		P2pReadBufferSize:             8 * 1024 * 1024,
+		IncomingIsLongest:             false,
 	}
 }
 

config/example_config.yaml

@@ -106,6 +106,7 @@ blocktx:
   registerTxsInterval: 10s # time interval to read from the channel registered transactions
   maxBlockProcessingDuration: 5m # maximum time a blocktx can spend on processing a block before unlocking it to be requested again
   monitorPeers: false # if enabled, peers which do not receive alive signal from nodes will be restarted
+  incomingIsLongest: false # whether each new block received is considered to be from the longest blockchain. If there are a lot of block gaps in blocktx database it is advisable to set this to true
   fillGaps:
     enabled: true
     interval: 15m # time interval to check and fill gaps in processed blocks

doc/README.md

@@ -58,12 +58,13 @@ stateDiagram-v2
     ACCEPTED_BY_NETWORK --> SEEN_ON_NETWORK: ARC has received Transaction ID\n announcement from another peer
     ACCEPTED_BY_NETWORK --> SEEN_IN_ORPHAN_MEMPOOL: Peer has sent a 'missing inputs' message
     SEEN_IN_ORPHAN_MEMPOOL --> SEEN_ON_NETWORK: All parent transactions\n have been received by peer
-    SEEN_ON_NETWORK --> MINED: Transaction ID was included in a BLOCK message
     SEEN_ON_NETWORK --> DOUBLE_SPEND_ATTEMPTED: A competing transactions entered the mempool
-    DOUBLE_SPEND_ATTEMPTED --> MINED: This transaction was accepted and mined
     DOUBLE_SPEND_ATTEMPTED --> REJECTED: This transaction was rejected in favor\n of one of the competing transactions
-    MINED --> MINED_IN_STALE_BLOCK: This transaction was mined in a block that became stale after reorg
+    DOUBLE_SPEND_ATTEMPTED --> MINED: This transaction was accepted and mined
+    SEEN_ON_NETWORK --> MINED: Transaction ID was included in a BLOCK message
+    MINED_IN_STALE_BLOCK --> SEEN_ON_NETWORK: Transaction gets re-broadcasted
     MINED --> [*]
+    MINED --> MINED_IN_STALE_BLOCK: This transaction was mined in a block that became stale after reorg
 ```
 
 ## Microservices

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0
+	github.com/jackc/pgx/v5 v5.3.1
 	github.com/jedib0t/go-pretty/v6 v6.5.9
 	github.com/jmoiron/sqlx v1.3.5
 	github.com/labstack/echo-contrib v0.17.1
@@ -99,6 +100,8 @@ require (
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/invopop/yaml v0.3.1 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect

go.sum

@@ -171,6 +171,8 @@ github.com/jackc/pgtype v1.14.0 h1:y+xUdabmyMkJLyApYuPj38mW+aAIqCe5uuBB51rH3Vw=
 github.com/jackc/pgtype v1.14.0/go.mod h1:LUMuVrfsFfdKGLw+AFFVv6KtHOFMwRgDDzBt76IqCA4=
 github.com/jackc/pgx/v4 v4.18.1 h1:YP7G1KABtKpB5IHrO9vYwSrCOhs7p3uqhvhhQBptya0=
 github.com/jackc/pgx/v4 v4.18.1/go.mod h1:FydWkUyadDmdNH/mHnGob881GawxeEm7TcMCzkb+qQE=
+github.com/jackc/pgx/v5 v5.3.1 h1:Fcr8QJ1ZeLi5zsPZqQeUZhNhxfkkKBOgJuYkJHoBOtU=
+github.com/jackc/pgx/v5 v5.3.1/go.mod h1:t3JDKnCBlYIc0ewLF0Q7B8MXmoIaBOZj/ic7iHozM/8=
 github.com/jedib0t/go-pretty/v6 v6.5.9 h1:ACteMBRrrmm1gMsXe9PSTOClQ63IXDUt03H5U+UV8OU=
 github.com/jedib0t/go-pretty/v6 v6.5.9/go.mod h1:zbn98qrYlh95FIhwwsbIip0LYpwSG8SUOScs+v9/t0E=
 github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g=

internal/blocktx/background_workers_test.go

@@ -8,14 +8,16 @@ import (
 	"testing"
 	"time"
 
+	"github.com/libsv/go-p2p"
+
 	"github.com/bitcoin-sv/arc/internal/blocktx"
 	"github.com/bitcoin-sv/arc/internal/blocktx/mocks"
 	"github.com/bitcoin-sv/arc/internal/blocktx/store"
 	storeMocks "github.com/bitcoin-sv/arc/internal/blocktx/store/mocks"
-	"github.com/libsv/go-p2p"
 
-	"github.com/bitcoin-sv/arc/internal/testdata"
 	"github.com/stretchr/testify/require"
+
+	"github.com/bitcoin-sv/arc/internal/testdata"
 )
 
 func TestStartFillGaps(t *testing.T) {

internal/blocktx/blocktx_api/blocktx_api.proto

@@ -11,14 +11,11 @@ service BlockTxAPI {
   // Health returns the health of the API.
   rpc Health (google.protobuf.Empty) returns (HealthResponse) {}
 
-  // ClearTransactions clears transaction data
-  rpc ClearTransactions(ClearData) returns (RowsAffectedResponse) {}
-
   // ClearBlocks clears block data
   rpc ClearBlocks(ClearData) returns (RowsAffectedResponse) {}
 
-  // ClearBlockTransactionsMap clears block-transaction-map data
-  rpc ClearBlockTransactionsMap(ClearData) returns (RowsAffectedResponse) {}
+  // ClearRegisteredTransactions clears registered transactions
+  rpc ClearRegisteredTransactions(ClearData) returns (RowsAffectedResponse) {}
 
   // DelUnfinishedBlockProcessing deletes unfinished block processing
   rpc DelUnfinishedBlockProcessing(DelUnfinishedBlockProcessingRequest) returns (RowsAffectedResponse) {}