sweepbatcher: allow swap_hash to be non-unique

SwapHash used to be a key in the sweeps table. Now the key is outpoint which
replaces columns outpoint_txid and outpoint_index. In-memory structures and unit
tests were also updated to use outpoint as key. Outpoint is truly unique.

Author: starius

loopdb/postgres.go

@@ -64,6 +64,21 @@ type PostgresStore struct {
 	*BaseDB
 }
 
+// In migration of sweeps table from outpoint_txid and outpoint_index to
+// outpoint we need to reverse the order of bytes in outpoint_txid and to
+// convert it to hex. This is done differently in sqlite and postgres.
+//
+// Changes from sqlite to postgres:
+//   - substr(blob, ...) -> get_byte(blob, index)
+//   - group_concat -> string_agg
+//   - 1-based indexing (32+1-i) -> 0-based (32 - i)
+//   - to_hex() + lpad(..., 2, '0') ensures each byte is two-digit hex
+const (
+	txidSqlite   = "group_concat(hex(substr(outpoint_txid,32+1-i,1)),'')"
+	txidPostgres = "string_agg(lpad(to_hex(get_byte(outpoint_txid, " +
+		"32 - i)), 2, '0'), '')"
+)
+
 // NewPostgresStore creates a new store that is backed by a Postgres database
 // backend.
 func NewPostgresStore(cfg *PostgresConfig,
@@ -93,6 +108,7 @@ func NewPostgresStore(cfg *PostgresConfig,
 		postgresFS := newReplacerFS(sqlSchemas, map[string]string{
 			"BLOB":                "BYTEA",
 			"INTEGER PRIMARY KEY": "SERIAL PRIMARY KEY",
+			txidSqlite:            txidPostgres,
 		})
 
 		err = applyMigrations(

loopdb/sqlc/batch.sql.go

@@ -0,0 +1,3 @@
+-- We kept old table as sweeps_old. Use it.
+ALTER TABLE sweeps RENAME TO sweeps_new;
+ALTER TABLE sweeps_old RENAME TO sweeps;

loopdb/sqlc/migrations/000013_batcher_key_outpoint.down.sql

@@ -0,0 +1,67 @@
+-- We want to make column swap_hash non-unique and to use the outpoint as a key.
+-- We can't make a column non-unique or remove it in sqlite, so work around.
+-- See https://stackoverflow.com/a/42013422
+
+-- We also made outpoint a single point replacing columns outpoint_txid and
+-- outpoint_index.
+
+-- sweeps stores the individual sweeps that are part of a batch.
+CREATE TABLE sweeps2 (
+        -- id is the autoincrementing primary key.
+        id INTEGER PRIMARY KEY,
+
+        -- swap_hash is the hash of the swap that is being swept.
+        swap_hash BLOB NOT NULL,
+
+        -- batch_id is the id of the batch this swap is part of.
+        batch_id INTEGER NOT NULL,
+
+        -- outpoint is the UTXO id of the output being swept ("txid:index").
+        outpoint TEXT NOT NULL UNIQUE,
+
+        -- amt is the amount of the output being swept.
+        amt BIGINT NOT NULL,
+
+        -- completed indicates whether the sweep has been completed.
+        completed BOOLEAN NOT NULL DEFAULT FALSE,
+
+        -- Foreign key constraint to ensure that we reference an existing batch
+        -- id.
+        FOREIGN KEY (batch_id) REFERENCES sweep_batches(id),
+
+        -- Foreign key constraint to ensure that swap_hash references an
+        -- existing swap.
+        FOREIGN KEY (swap_hash) REFERENCES swaps(swap_hash)
+);
+
+-- Copy all the data from sweeps to sweeps2.
+-- Explanation:
+-- - seq(i) goes from 1 to 32
+-- - substr(outpoint_txid, 32+1-i, 1) indexes BLOB bytes in reverse order
+--   (SQLite uses 1-based indexing)
+-- - hex(...) gives uppercase by default, so wrapped in lower(...)
+-- - group_concat(..., '') combines all hex digits
+-- - concatenated with ':' || CAST(outpoint_index AS TEXT) for full outpoint.
+WITH RECURSIVE seq(i) AS (
+  SELECT 1
+  UNION ALL
+  SELECT i + 1 FROM seq WHERE i < 32
+)
+INSERT INTO sweeps2 (
+    id, swap_hash, batch_id, outpoint, amt, completed
+)
+SELECT
+    id,
+    swap_hash,
+    batch_id,
+    (
+      SELECT lower(group_concat(hex(substr(outpoint_txid,32+1-i,1)),''))
+      FROM seq
+    ) || ':' || CAST(outpoint_index AS TEXT),
+    amt,
+    completed
+FROM sweeps;
+
+-- Rename tables.
+ALTER TABLE sweeps RENAME TO sweeps_old;
+ALTER TABLE sweeps2 RENAME TO sweeps;

loopdb/sqlc/migrations/000013_batcher_key_outpoint.up.sql

@@ -47,23 +47,18 @@ WHERE
 INSERT INTO sweeps (
         swap_hash,
         batch_id,
-        outpoint_txid,
-        outpoint_index,
+        outpoint,
         amt,
         completed
 ) VALUES (
         $1,
         $2,
         $3,
         $4,
-        $5,
-        $6
-) ON CONFLICT (swap_hash) DO UPDATE SET
+        $5
+) ON CONFLICT (outpoint) DO UPDATE SET
         batch_id = $2,
-        outpoint_txid = $3,
-        outpoint_index = $4,
-        amt = $5,
-        completed = $6;
+        completed = $5;
 
 -- name: GetParentBatch :one
 SELECT
@@ -73,7 +68,7 @@ FROM
 JOIN
         sweeps ON sweep_batches.id = sweeps.batch_id
 WHERE
-        sweeps.swap_hash = $1;
+        sweeps.outpoint = $1;
 
 -- name: GetBatchSweptAmount :one
 SELECT
@@ -101,4 +96,4 @@ SELECT
 FROM
     (SELECT false AS false_value) AS f
 LEFT JOIN
-    sweeps s ON s.swap_hash = $1;
+    sweeps s ON s.outpoint = $1;

loopdb/sqlc/models.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
 	sweeppkg "github.com/lightninglabs/loop/sweep"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/lntypes"
@@ -108,8 +109,8 @@ func estimateSweepFeeIncrement(s *sweep) (feeDetails, feeDetails, error) {
 		rbfCache: rbfCache{
 			FeeRate: s.minFeeRate,
 		},
-		sweeps: map[lntypes.Hash]sweep{
-			s.swapHash: *s,
+		sweeps: map[wire.OutPoint]sweep{
+			s.outpoint: *s,
 		},
 	}
 
@@ -120,9 +121,13 @@ func estimateSweepFeeIncrement(s *sweep) (feeDetails, feeDetails, error) {
 	}
 
 	// Add the same sweep again to measure weight increments.
-	swapHash2 := s.swapHash
-	swapHash2[0]++
-	batch.sweeps[swapHash2] = *s
+	outpoint2 := s.outpoint
+	outpoint2.Hash[0]++
+	if _, has := batch.sweeps[outpoint2]; has {
+		return feeDetails{}, feeDetails{}, fmt.Errorf("dummy outpoint "+
+			"%s is present in the batch", outpoint2)
+	}
+	batch.sweeps[outpoint2] = *s
 
 	// Estimate weight of a batch with two sweeps.
 	fd2, err := estimateBatchWeight(batch)