Reapply 8644 on 9260

.github/workflows/main.yml

@@ -284,7 +284,7 @@ jobs:
         run: ./scripts/install_bitcoind.sh $BITCOIN_VERSION
 
       - name: run ${{ matrix.name }}
-        run: make itest-parallel tranches=${{ env.TRANCHES }} ${{ matrix.args }}
+        run: make itest-parallel tranches=${{ env.TRANCHES }} ${{ matrix.args }} shuffleseed=${{ github.run_id }}${{ strategy.job-index }}
 
       - name: Send coverage
         if: ${{ contains(matrix.args, 'cover=1') }}
@@ -297,7 +297,7 @@ jobs:
       - name: Zip log files on failure
         if: ${{ failure() }}
         timeout-minutes: 5 # timeout after 5 minute
-        run: 7z a logs-itest-${{ matrix.name }}.zip itest/**/*.log
+        run: 7z a logs-itest-${{ matrix.name }}.zip itest/**/*.log itest/postgres.log
 
       - name: Upload log files on failure
         uses: actions/upload-artifact@v3
@@ -332,7 +332,7 @@ jobs:
           key-prefix: integration-test
 
       - name: run itest
-        run: make itest-parallel tranches=${{ env.TRANCHES }} windows=1
+        run: make itest-parallel tranches=${{ env.TRANCHES }} windows=1 shuffleseed=${{ github.run_id }}
 
       - name: kill any remaining lnd processes
         if: ${{ failure() }}
@@ -375,14 +375,8 @@ jobs:
           go-version: '${{ env.GO_VERSION }}'
           key-prefix: integration-test
 
-      - name: install bitcoind
-        run: |
-          wget https://bitcoincore.org/bin/bitcoin-core-${BITCOIN_VERSION}.0/bitcoin-${BITCOIN_VERSION}.0-arm64-apple-darwin.tar.gz
-          tar zxvf bitcoin-${BITCOIN_VERSION}.0-arm64-apple-darwin.tar.gz
-          mv bitcoin-${BITCOIN_VERSION}.0 /tmp/bitcoin
-
       - name: run itest
-        run: PATH=$PATH:/tmp/bitcoin/bin make itest-parallel tranches=${{ env.TRANCHES }} backend=bitcoind
+        run: PATH=$PATH:/tmp/bitcoin/bin make itest-parallel tranches=${{ env.TRANCHES }} shuffleseed=${{ github.run_id }}
 
       - name: Zip log files on failure
         if: ${{ failure() }}

Makefile

@@ -195,20 +195,25 @@ ifeq ($(dbbackend),postgres)
 	docker rm lnd-postgres --force || echo "Starting new postgres container"
 
 	# Start a fresh postgres instance. Allow a maximum of 500 connections so
-	# that multiple lnd instances with a maximum number of connections of 50
-	# each can run concurrently.
-	docker run --name lnd-postgres -e POSTGRES_PASSWORD=postgres -p 6432:5432 -d postgres:13-alpine -N 500
-	docker logs -f lnd-postgres &
+	# that multiple lnd instances with a maximum number of connections of 20
+	# each can run concurrently. Note that many of the settings here are
+	# specifically for integration testing and are not fit for running
+	# production nodes.
+	docker run --name lnd-postgres -e POSTGRES_PASSWORD=postgres -p 6432:5432 -d postgres:13-alpine -N 1500 -c max_pred_locks_per_transaction=1024 -c max_locks_per_transaction=128 -c jit=off -c work_mem=8MB -c checkpoint_timeout=10min -c enable_seqscan=off
+	docker logs -f lnd-postgres >itest/postgres.log 2>&1 &
 
 	# Wait for the instance to be started.
 	sleep $(POSTGRES_START_DELAY)
 endif
 
+clean-itest-logs:
+	rm -rf itest/*.log itest/.logs-*
+
 #? itest-only: Only run integration tests without re-building binaries
-itest-only: db-instance
+itest-only: clean-itest-logs db-instance
 	@$(call print, "Running integration tests with ${backend} backend.")
-	rm -rf itest/*.log itest/.logs-*; date
-	EXEC_SUFFIX=$(EXEC_SUFFIX) scripts/itest_part.sh 0 1 $(TEST_FLAGS) $(ITEST_FLAGS) -test.v
+	date
+	EXEC_SUFFIX=$(EXEC_SUFFIX) scripts/itest_part.sh 0 1 $(SHUFFLE_SEED) $(TEST_FLAGS) $(ITEST_FLAGS) -test.v
 	$(COLLECT_ITEST_COVERAGE)
 
 #? itest: Build and run integration tests
@@ -218,10 +223,10 @@ itest: build-itest itest-only
 itest-race: build-itest-race itest-only
 
 #? itest-parallel: Build and run integration tests in parallel mode, running up to ITEST_PARALLELISM test tranches in parallel (default 4)
-itest-parallel: build-itest db-instance
+itest-parallel: clean-itest-logs build-itest db-instance
 	@$(call print, "Running tests")
-	rm -rf itest/*.log itest/.logs-*; date
-	EXEC_SUFFIX=$(EXEC_SUFFIX) scripts/itest_parallel.sh $(ITEST_PARALLELISM) $(NUM_ITEST_TRANCHES) $(TEST_FLAGS) $(ITEST_FLAGS)
+	date
+	EXEC_SUFFIX=$(EXEC_SUFFIX) scripts/itest_parallel.sh $(ITEST_PARALLELISM) $(NUM_ITEST_TRANCHES) $(SHUFFLE_SEED) $(TEST_FLAGS) $(ITEST_FLAGS)
 	$(COLLECT_ITEST_COVERAGE)
 
 #? itest-clean: Kill all running itest processes

batch/batch.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/sqldb"
 )
 
 // errSolo is a sentinel error indicating that the requester should re-run the
@@ -55,7 +56,19 @@ func (b *batch) run() {
 			for i, req := range b.reqs {
 				err := req.Update(tx)
 				if err != nil {
-					failIdx = i
+					// If we get a serialization error, we
+					// want the underlying SQL retry
+					// mechanism to retry the entire batch.
+					// Otherwise, we can succeed in an
+					// sqldb retry and still re-execute the
+					// failing request individually.
+					dbErr := sqldb.MapSQLError(err)
+					if !sqldb.IsSerializationError(dbErr) {
+						failIdx = i
+
+						return dbErr
+					}
+
 					return err
 				}
 			}

chainio/README.md

@@ -0,0 +1,152 @@
+# Chainio 
+
+`chainio` is a package designed to provide blockchain data access to various
+subsystems within `lnd`. When a new block is received, it is encapsulated in a
+`Blockbeat` object and disseminated to all registered consumers. Consumers may
+receive these updates either concurrently or sequentially, based on their
+registration configuration, ensuring that each subsystem maintains a
+synchronized view of the current block state.
+
+The main components include:
+
+- `Blockbeat`: An interface that provides information about the block.
+
+- `Consumer`: An interface that specifies how subsystems handle the blockbeat.
+
+- `BlockbeatDispatcher`: The core service responsible for receiving each block
+  and distributing it to all consumers.
+
+Additionally, the `BeatConsumer` struct provides a partial implementation of
+the `Consumer` interface. This struct helps reduce code duplication, allowing
+subsystems to avoid re-implementing the `ProcessBlock` method and provides a
+commonly used `NotifyBlockProcessed` method.
+
+
+### Register a Consumer
+
+Consumers within the same queue are notified **sequentially**, while all queues
+are notified **concurrently**. A queue consists of a slice of consumers, which
+are notified in left-to-right order. Developers are responsible for determining
+dependencies in block consumption across subsystems: independent subsystems
+should be notified concurrently, whereas dependent subsystems should be
+notified sequentially.
+
+To notify the consumers concurrently, put them in different queues,
+```go
+// consumer1 and consumer2 will be notified concurrently.
+queue1 := []chainio.Consumer{consumer1}
+blockbeatDispatcher.RegisterQueue(consumer1)
+
+queue2 := []chainio.Consumer{consumer2}
+blockbeatDispatcher.RegisterQueue(consumer2)
+```
+
+To notify the consumers sequentially, put them in the same queue,
+```go
+// consumers will be notified sequentially via,
+// consumer1 -> consumer2 -> consumer3
+queue := []chainio.Consumer{
+   consumer1,
+   consumer2,
+   consumer3,
+}
+blockbeatDispatcher.RegisterQueue(queue)
+```
+
+### Implement the `Consumer` Interface
+
+Implementing the `Consumer` interface is straightforward. Below is an example
+of how
+[`sweep.TxPublisher`](https://github.com/lightningnetwork/lnd/blob/5cec466fad44c582a64cfaeb91f6d5fd302fcf85/sweep/fee_bumper.go#L310)
+implements this interface.
+
+To start, embed the partial implementation `chainio.BeatConsumer`, which
+already provides the `ProcessBlock` implementation and commonly used
+`NotifyBlockProcessed` method,  and exposes `BlockbeatChan` for the consumer to
+receive blockbeats.
+
+```go
+type TxPublisher struct {
+   started atomic.Bool
+   stopped atomic.Bool
+
+   chainio.BeatConsumer
+
+   ...
+```
+
+We should also remember to initialize this `BeatConsumer`,
+
+```go
+...
+// Mount the block consumer.
+tp.BeatConsumer = chainio.NewBeatConsumer(tp.quit, tp.Name())
+```
+
+Finally, in the main event loop, read from `BlockbeatChan`, process the
+received blockbeat, and, crucially, call `tp.NotifyBlockProcessed` to inform
+the blockbeat dispatcher that processing is complete.
+
+```go
+for {
+      select {
+      case beat := <-tp.BlockbeatChan:
+         // Consume this blockbeat, usually it means updating the subsystem
+         // using the new block data.
+
+         // Notify we've processed the block.
+         tp.NotifyBlockProcessed(beat, nil)
+
+      ...
+```
+
+### Existing Queues
+
+Currently, we have a single queue of consumers dedicated to handling force
+closures. This queue includes `ChainArbitrator`, `UtxoSweeper`, and
+`TxPublisher`, with `ChainArbitrator` managing two internal consumers:
+`chainWatcher` and `ChannelArbitrator`. The blockbeat flows sequentially
+through the chain as follows: `ChainArbitrator => chainWatcher =>
+ChannelArbitrator => UtxoSweeper => TxPublisher`. The following diagram
+illustrates the flow within the public subsystems.
+
+```mermaid
+sequenceDiagram
+		autonumber
+		participant bb as BlockBeat
+		participant cc as ChainArb
+		participant us as UtxoSweeper
+		participant tp as TxPublisher
+
+		note left of bb: 0. received block x,<br>dispatching...
+
+    note over bb,cc: 1. send block x to ChainArb,<br>wait for its done signal
+		bb->>cc: block x
+		rect rgba(165, 0, 85, 0.8)
+      critical signal processed
+        cc->>bb: processed block
+      option Process error or timeout
+        bb->>bb: error and exit
+      end
+    end
+
+    note over bb,us: 2. send block x to UtxoSweeper, wait for its done signal
+		bb->>us: block x
+		rect rgba(165, 0, 85, 0.8)
+      critical signal processed
+        us->>bb: processed block
+      option Process error or timeout
+        bb->>bb: error and exit
+      end
+    end
+
+    note over bb,tp: 3. send block x to TxPublisher, wait for its done signal
+		bb->>tp: block x
+		rect rgba(165, 0, 85, 0.8)
+      critical signal processed
+        tp->>bb: processed block
+      option Process error or timeout
+        bb->>bb: error and exit
+      end
+    end
+```

chainio/blockbeat.go

@@ -0,0 +1,55 @@
+package chainio
+
+import (
+	"fmt"
+
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/build"
+	"github.com/lightningnetwork/lnd/chainntnfs"
+)
+
+// Beat implements the Blockbeat interface. It contains the block epoch and a
+// customized logger.
+//
+// TODO(yy): extend this to check for confirmation status - which serves as the
+// single source of truth, to avoid the potential race between receiving blocks
+// and `GetTransactionDetails/RegisterSpendNtfn/RegisterConfirmationsNtfn`.
+type Beat struct {
+	// epoch is the current block epoch the blockbeat is aware of.
+	epoch chainntnfs.BlockEpoch
+
+	// log is the customized logger for the blockbeat which prints the
+	// block height.
+	log btclog.Logger
+}
+
+// Compile-time check to ensure Beat satisfies the Blockbeat interface.
+var _ Blockbeat = (*Beat)(nil)
+
+// NewBeat creates a new beat with the specified block epoch and a customized
+// logger.
+func NewBeat(epoch chainntnfs.BlockEpoch) *Beat {
+	b := &Beat{
+		epoch: epoch,
+	}
+
+	// Create a customized logger for the blockbeat.
+	logPrefix := fmt.Sprintf("Height[%6d]:", b.Height())
+	b.log = build.NewPrefixLog(logPrefix, clog)
+
+	return b
+}
+
+// Height returns the height of the block epoch.
+//
+// NOTE: Part of the Blockbeat interface.
+func (b *Beat) Height() int32 {
+	return b.epoch.Height
+}
+
+// logger returns the logger for the blockbeat.
+//
+// NOTE: Part of the private blockbeat interface.
+func (b *Beat) logger() btclog.Logger {
+	return b.log
+}

chainio/blockbeat_test.go

@@ -0,0 +1,28 @@
+package chainio
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/stretchr/testify/require"
+)
+
+var errDummy = errors.New("dummy error")
+
+// TestNewBeat tests the NewBeat and Height functions.
+func TestNewBeat(t *testing.T) {
+	t.Parallel()
+
+	// Create a testing epoch.
+	epoch := chainntnfs.BlockEpoch{
+		Height: 1,
+	}
+
+	// Create the beat and check the internal state.
+	beat := NewBeat(epoch)
+	require.Equal(t, epoch, beat.epoch)
+
+	// Check the height function.
+	require.Equal(t, epoch.Height, beat.Height())
+}