Fix: postgres pay race condition (#1222)

* chore: add test for postgres race condition in transaction service * fix: payments race condition when using postgres backend * chore: reduce delay in concurrent payment tests
2026-08-13 12:33:39 +02:00 · 2025-03-25 17:31:19 +07:00 · 2025-03-25 17:31:19 +07:00 · 150d7bcc7a
commit 150d7bcc7a
parent c6a2125a1a
5 changed files with 304 additions and 68 deletions
--- a/README.md
+++ b/README.md
@ -95,6 +95,22 @@ _If you get a blank screen, try running in your normal terminal (outside of vsco

    $ go test ./... -run TestHandleGetInfoEvent

+#### Testing with PostgreSQL
+
+By default, sqlite is used for testing. It is also possible to run the tests with PostgreSQL.
+
+The tests use [pgtestdb](https://github.com/peterldowns/pgtestdb) to set up a temporary PostgreSQL database, which requires a running PostgreSQL server. Follow your OS instructions to install PostgreSQL, or use the official [Docker image](https://hub.docker.com/_/postgres).
+
+See the [docker compose file](./tests/db/postgres/docker-compose.yml) for an easy way to get started.
+
+When PostgreSQL is installed and running, set the `TEST_DATABASE_URI` environment variable to the PostgreSQL connection string. For example:
+
+    $ export TEST_DATABASE_URI="postgresql://user:password@localhost:5432/postgres"
+
+Note that the PostgreSQL user account must be granted appropriate permissions to create new databases. When the tests complete, the temporary database will be removed.
+
+**Do not** use a production database. It is preferable to launch a dedicated PostgreSQL instance for testing purposes.
+
 #### Mocking

 We use [testify/mock](https://github.com/stretchr/testify) to facilitate mocking in tests. Instead of writing mocks manually, we generate them using [vektra/mockery](https://github.com/vektra/mockery). To regenerate them, [install mockery](https://vektra.github.io/mockery/latest/installation) and run it in the project's root directory:
--- a/nip47/controllers/multi_pay_invoice_controller_test.go
+++ b/nip47/controllers/multi_pay_invoice_controller_test.go
@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"slices"
 	"sync"
 	"testing"

@ -421,3 +422,91 @@ func TestHandleMultiPayInvoiceEvent_LNClient_OnePaymentFailed(t *testing.T) {
 	assert.Equal(t, constants.ERROR_INTERNAL, responses[1].Error.Code)
 	assert.Equal(t, "Some error", responses[1].Error.Message)
 }
+
+func TestHandleMultiPayInvoiceEvent_IsolatedApp_ConcurrentPayments(t *testing.T) {
+	ctx := context.TODO()
+
+	svc, err := tests.CreateTestService(t)
+	require.NoError(t, err)
+	defer svc.Remove()
+
+	app, _, err := tests.CreateApp(svc)
+	assert.NoError(t, err)
+	app.Isolated = true
+	svc.DB.Save(&app)
+
+	svc.DB.Create(&db.Transaction{
+		AppId: &app.ID,
+		State: constants.TRANSACTION_STATE_SETTLED,
+		Type:  constants.TRANSACTION_TYPE_INCOMING,
+		// invoices paid are 123000 millisats
+		AmountMsat: 200000,
+	})
+
+	// force delay inside transaction
+	if svc.DB.Dialector.Name() == "postgres" {
+		err = svc.DB.Exec(`
+CREATE OR REPLACE FUNCTION slow_down_query()
+RETURNS TRIGGER AS $slow_down_query$
+BEGIN
+    -- Introduce a delay of 1 second
+    PERFORM pg_sleep(1);
+    RETURN NEW;
+END;
+$slow_down_query$ LANGUAGE plpgsql;
+
+CREATE TRIGGER slow_down_query
+AFTER INSERT ON transactions
+FOR EACH ROW
+EXECUTE PROCEDURE slow_down_query();`).Error
+
+		require.NoError(t, err)
+	}
+
+	appPermission := &db.AppPermission{
+		AppId: app.ID,
+		App:   *app,
+		Scope: constants.PAY_INVOICE_SCOPE,
+	}
+	err = svc.DB.Create(appPermission).Error
+	assert.NoError(t, err)
+
+	nip47Request := &models.Request{}
+	err = json.Unmarshal([]byte(nip47MultiPayJson), nip47Request)
+	assert.NoError(t, err)
+
+	responses := []*models.Response{}
+
+	var mu sync.Mutex
+
+	publishResponse := func(response *models.Response, tags nostr.Tags) {
+		mu.Lock()
+		defer mu.Unlock()
+		responses = append(responses, response)
+	}
+
+	dbRequestEvent := &db.RequestEvent{}
+	err = svc.DB.Create(&dbRequestEvent).Error
+	assert.NoError(t, err)
+
+	NewTestNip47Controller(svc).
+		HandleMultiPayInvoiceEvent(ctx, nip47Request, dbRequestEvent.ID, app, publishResponse)
+
+	require.Equal(t, 2, len(responses))
+
+	// we can't guarantee which request was processed first
+	// so put the successful one at the front
+	successfulIdx := slices.IndexFunc(responses, func(r *models.Response) bool {
+		return r.Result != nil
+	})
+	require.GreaterOrEqual(t, successfulIdx, 0)
+
+	if successfulIdx > 0 {
+		responses[0], responses[successfulIdx] = responses[successfulIdx], responses[0]
+	}
+
+	for _, response := range responses[1:] {
+		require.Nil(t, response.Result)
+		assert.Equal(t, constants.ERROR_INSUFFICIENT_BALANCE, response.Error.Code)
+	}
+}
--- a/nip47/controllers/multi_pay_keysend_controller_test.go
+++ b/nip47/controllers/multi_pay_keysend_controller_test.go
@ -3,6 +3,7 @@ package controllers
 import (
 	"context"
 	"encoding/json"
+	"slices"
 	"sync"
 	"testing"

@ -172,3 +173,94 @@ func TestHandleMultiPayKeysendEvent_OneBudgetExceeded(t *testing.T) {
 	assert.Nil(t, responses[1].Result)
 	assert.Equal(t, constants.ERROR_QUOTA_EXCEEDED, responses[1].Error.Code)
 }
+
+func TestHandleMultiPayKeysendEvent_IsolatedApp_ConcurrentPayments(t *testing.T) {
+	ctx := context.TODO()
+	svc, err := tests.CreateTestService(t)
+	require.NoError(t, err)
+	defer svc.Remove()
+
+	app, _, err := tests.CreateApp(svc)
+	app.Isolated = true
+	assert.NoError(t, err)
+	app.Isolated = true
+	svc.DB.Save(&app)
+
+	appPermission := &db.AppPermission{
+		AppId:        app.ID,
+		App:          *app,
+		Scope:        constants.PAY_INVOICE_SCOPE,
+		MaxAmountSat: 400,
+	}
+	err = svc.DB.Create(appPermission).Error
+	assert.NoError(t, err)
+
+	svc.DB.Create(&db.Transaction{
+		AppId: &app.ID,
+		State: constants.TRANSACTION_STATE_SETTLED,
+		Type:  constants.TRANSACTION_TYPE_INCOMING,
+		// keysends paid are 123000 millisats
+		AmountMsat: 200000,
+	})
+
+	// force delay inside transaction
+	if svc.DB.Dialector.Name() == "postgres" {
+		err = svc.DB.Exec(`
+CREATE OR REPLACE FUNCTION slow_down_query()
+RETURNS TRIGGER AS $slow_down_query$
+BEGIN
+    -- Introduce a delay of 1 second
+    PERFORM pg_sleep(1);
+    RETURN NEW;
+END;
+$slow_down_query$ LANGUAGE plpgsql;
+
+CREATE TRIGGER slow_down_query
+AFTER INSERT ON transactions
+FOR EACH ROW
+EXECUTE PROCEDURE slow_down_query();`).Error
+
+		require.NoError(t, err)
+	}
+
+	nip47Request := &models.Request{}
+	err = json.Unmarshal([]byte(nip47MultiPayKeysendJson), nip47Request)
+	assert.NoError(t, err)
+
+	dbRequestEvent := &db.RequestEvent{}
+	err = svc.DB.Create(&dbRequestEvent).Error
+	assert.NoError(t, err)
+
+	responses := []*models.Response{}
+	dTags := []nostr.Tags{}
+
+	var mu sync.Mutex
+
+	publishResponse := func(response *models.Response, tags nostr.Tags) {
+		mu.Lock()
+		defer mu.Unlock()
+		responses = append(responses, response)
+		dTags = append(dTags, tags)
+	}
+
+	NewTestNip47Controller(svc).
+		HandleMultiPayKeysendEvent(ctx, nip47Request, dbRequestEvent.ID, app, publishResponse)
+
+	require.Equal(t, 2, len(responses))
+
+	// we can't guarantee which request was processed first
+	// so put the successful one at the front
+	successfulIdx := slices.IndexFunc(responses, func(r *models.Response) bool {
+		return r.Result != nil
+	})
+	require.GreaterOrEqual(t, successfulIdx, 0)
+
+	if successfulIdx > 0 {
+		responses[0], responses[successfulIdx] = responses[successfulIdx], responses[0]
+	}
+
+	for _, response := range responses[1:] {
+		require.Nil(t, response.Result)
+		assert.Equal(t, constants.ERROR_INSUFFICIENT_BALANCE, response.Error.Code)
+	}
+}
--- a/tests/db/postgres/docker-compose.yml
+++ b/tests/db/postgres/docker-compose.yml
@ -0,0 +1,26 @@
+# docker compose up
+# connect with psql postgresql://postgres:password@localhost:5434
+# TEST_DATABASE_URI="postgresql://postgres:password@localhost:5434" go test -timeout 30s -run ^TestHandleMultiPayInvoiceEvent_IsolatedApp_ConcurrentPayments$ github.com/getAlby/hub/nip47/controllers
+# or
+# TEST_DATABASE_URI="postgresql://postgres:password@localhost:5434" go test -timeout 30s -run ^TestHandleMultiPayKeysendEvent_IsolatedApp_ConcurrentPayments$ github.com/getAlby/hub/nip47/controllers
+version: "3.6"
+services:
+  pgtestdb:
+    image: postgres:15
+    environment:
+      POSTGRES_PASSWORD: password
+    restart: unless-stopped
+    volumes:
+      # Uses a tmpfs volume to make tests extremely fast. The data in test
+      # databases is not persisted across restarts, nor does it need to be.
+      - type: tmpfs
+        target: /var/lib/postgresql/data/
+    command:
+      - "postgres"
+      - "-c" # turn off fsync for speed
+      - "fsync=off"
+      - "-c" # log everything for debugging
+      - "log_statement=all"
+    ports:
+      # Entirely up to you what port you want to use while testing.
+      - "5434:5432"
--- a/transactions/transactions_service.go
+++ b/transactions/transactions_service.go
@ -12,6 +12,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	decodepay "github.com/nbd-wtf/ln-decodepay"
@ -47,6 +48,10 @@ const (
 	CustomKeyTlvType  = 696969
 )

+// Prevent races when checking the current balance and creating payment
+// transactions from concurrent goroutines.
+var balanceValidationLock = &sync.Mutex{}
+
 type Transaction = db.Transaction

 type Boostagram struct {
@ -225,53 +230,57 @@ func (svc *transactionsService) SendPaymentSync(ctx context.Context, payReq stri
 		paymentAmount = *amountMsat
 	}

-	err = svc.db.Transaction(func(tx *gorm.DB) error {
-		var existingSettledTransaction db.Transaction
-		if tx.Limit(1).Find(&existingSettledTransaction, &db.Transaction{
-			Type:        constants.TRANSACTION_TYPE_OUTGOING,
-			PaymentHash: paymentRequest.PaymentHash,
-			State:       constants.TRANSACTION_STATE_SETTLED,
-		}).RowsAffected > 0 {
-			logger.Logger.WithField("payment_hash", dbTransaction.PaymentHash).Debug("this invoice has already been paid")
-			return errors.New("this invoice has already been paid")
-		}
-		if tx.Limit(1).Find(&existingSettledTransaction, &db.Transaction{
-			Type:        constants.TRANSACTION_TYPE_OUTGOING,
-			PaymentHash: paymentRequest.PaymentHash,
-			State:       constants.TRANSACTION_STATE_PENDING,
-		}).RowsAffected > 0 {
-			logger.Logger.WithField("payment_hash", dbTransaction.PaymentHash).Debug("this invoice is already being paid")
-			return errors.New("there is already a payment pending for this invoice")
-		}
+	err = func() error {
+		balanceValidationLock.Lock()
+		defer balanceValidationLock.Unlock()
+		return svc.db.Transaction(func(tx *gorm.DB) error {
+			var existingSettledTransaction db.Transaction
+			if tx.Limit(1).Find(&existingSettledTransaction, &db.Transaction{
+				Type:        constants.TRANSACTION_TYPE_OUTGOING,
+				PaymentHash: paymentRequest.PaymentHash,
+				State:       constants.TRANSACTION_STATE_SETTLED,
+			}).RowsAffected > 0 {
+				logger.Logger.WithField("payment_hash", dbTransaction.PaymentHash).Debug("this invoice has already been paid")
+				return errors.New("this invoice has already been paid")
+			}
+			if tx.Limit(1).Find(&existingSettledTransaction, &db.Transaction{
+				Type:        constants.TRANSACTION_TYPE_OUTGOING,
+				PaymentHash: paymentRequest.PaymentHash,
+				State:       constants.TRANSACTION_STATE_PENDING,
+			}).RowsAffected > 0 {
+				logger.Logger.WithField("payment_hash", dbTransaction.PaymentHash).Debug("this invoice is already being paid")
+				return errors.New("there is already a payment pending for this invoice")
+			}

-		err := svc.validateCanPay(tx, appId, paymentAmount, paymentRequest.Description)
-		if err != nil {
+			err := svc.validateCanPay(tx, appId, paymentAmount, paymentRequest.Description)
+			if err != nil {
+				return err
+			}
+
+			var expiresAt *time.Time
+			if paymentRequest.Expiry > 0 {
+				expiresAtValue := time.Now().Add(time.Duration(paymentRequest.Expiry) * time.Second)
+				expiresAt = &expiresAtValue
+			}
+			dbTransaction = db.Transaction{
+				AppId:           appId,
+				RequestEventId:  requestEventId,
+				Type:            constants.TRANSACTION_TYPE_OUTGOING,
+				State:           constants.TRANSACTION_STATE_PENDING,
+				FeeReserveMsat:  CalculateFeeReserveMsat(paymentAmount),
+				AmountMsat:      paymentAmount,
+				PaymentRequest:  payReq,
+				PaymentHash:     paymentRequest.PaymentHash,
+				Description:     paymentRequest.Description,
+				DescriptionHash: paymentRequest.DescriptionHash,
+				ExpiresAt:       expiresAt,
+				SelfPayment:     selfPayment,
+				Metadata:        datatypes.JSON(metadataBytes),
+			}
+			err = tx.Create(&dbTransaction).Error
 			return err
-		}
-
-		var expiresAt *time.Time
-		if paymentRequest.Expiry > 0 {
-			expiresAtValue := time.Now().Add(time.Duration(paymentRequest.Expiry) * time.Second)
-			expiresAt = &expiresAtValue
-		}
-		dbTransaction = db.Transaction{
-			AppId:           appId,
-			RequestEventId:  requestEventId,
-			Type:            constants.TRANSACTION_TYPE_OUTGOING,
-			State:           constants.TRANSACTION_STATE_PENDING,
-			FeeReserveMsat:  CalculateFeeReserveMsat(paymentAmount),
-			AmountMsat:      paymentAmount,
-			PaymentRequest:  payReq,
-			PaymentHash:     paymentRequest.PaymentHash,
-			Description:     paymentRequest.Description,
-			DescriptionHash: paymentRequest.DescriptionHash,
-			ExpiresAt:       expiresAt,
-			SelfPayment:     selfPayment,
-			Metadata:        datatypes.JSON(metadataBytes),
-		}
-		err = tx.Create(&dbTransaction).Error
-		return err
-	})
+		})
+	}()

 	if err != nil {
 		logger.Logger.WithFields(logrus.Fields{
@ -360,30 +369,34 @@ func (svc *transactionsService) SendKeysend(ctx context.Context, amount uint64,

 	selfPayment := destination == lnClient.GetPubkey()

-	err = svc.db.Transaction(func(tx *gorm.DB) error {
-		err := svc.validateCanPay(tx, appId, amount, "")
-		if err != nil {
+	err = func() error {
+		balanceValidationLock.Lock()
+		defer balanceValidationLock.Unlock()
+		return svc.db.Transaction(func(tx *gorm.DB) error {
+			err := svc.validateCanPay(tx, appId, amount, "")
+			if err != nil {
+				return err
+			}
+
+			dbTransaction = db.Transaction{
+				AppId:          appId,
+				Description:    svc.getDescriptionFromCustomRecords(customRecords),
+				RequestEventId: requestEventId,
+				Type:           constants.TRANSACTION_TYPE_OUTGOING,
+				State:          constants.TRANSACTION_STATE_PENDING,
+				FeeReserveMsat: CalculateFeeReserveMsat(uint64(amount)),
+				AmountMsat:     amount,
+				Metadata:       datatypes.JSON(metadataBytes),
+				Boostagram:     datatypes.JSON(boostagramBytes),
+				PaymentHash:    paymentHash,
+				Preimage:       &preimage,
+				SelfPayment:    selfPayment,
+			}
+			err = tx.Create(&dbTransaction).Error
+
 			return err
-		}
-
-		dbTransaction = db.Transaction{
-			AppId:          appId,
-			Description:    svc.getDescriptionFromCustomRecords(customRecords),
-			RequestEventId: requestEventId,
-			Type:           constants.TRANSACTION_TYPE_OUTGOING,
-			State:          constants.TRANSACTION_STATE_PENDING,
-			FeeReserveMsat: CalculateFeeReserveMsat(uint64(amount)),
-			AmountMsat:     amount,
-			Metadata:       datatypes.JSON(metadataBytes),
-			Boostagram:     datatypes.JSON(boostagramBytes),
-			PaymentHash:    paymentHash,
-			Preimage:       &preimage,
-			SelfPayment:    selfPayment,
-		}
-		err = tx.Create(&dbTransaction).Error
-
-		return err
-	})
+		})
+	}()

 	if err != nil {
 		logger.Logger.WithFields(logrus.Fields{