diff --git a/go.mod b/go.mod index bab5d103..5e8aeb43 100755 --- a/go.mod +++ b/go.mod @@ -31,10 +31,10 @@ require ( github.com/testcontainers/testcontainers-go/modules/postgres v0.33.0 github.com/yaacov/tree-search-language v0.0.0-20190923184055-1c2dad2e354b go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0 - go.opentelemetry.io/otel v1.38.0 + go.opentelemetry.io/otel v1.39.0 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.38.0 - go.opentelemetry.io/otel/sdk v1.38.0 - go.opentelemetry.io/otel/trace v1.38.0 + go.opentelemetry.io/otel/sdk v1.39.0 + go.opentelemetry.io/otel/trace v1.39.0 go.uber.org/mock v0.6.0 gopkg.in/resty.v1 v1.12.0 gorm.io/datatypes v1.2.7 @@ -133,15 +133,16 @@ require ( github.com/yusufpapurcu/wmi v1.2.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 // indirect - go.opentelemetry.io/otel/metric v1.38.0 // indirect + go.opentelemetry.io/otel/metric v1.39.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.46.0 // indirect - golang.org/x/net v0.47.0 // indirect + golang.org/x/net v0.48.0 // indirect golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.39.0 // indirect golang.org/x/text v0.32.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect + google.golang.org/grpc v1.79.3 // indirect google.golang.org/protobuf v1.36.10 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect gorm.io/driver/mysql v1.5.6 // indirect diff --git a/go.sum b/go.sum index c585349f..c6ba4d7f 100755 --- a/go.sum +++ b/go.sum @@ -642,22 +642,22 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0 h1:Hf9xI/XLML9ElpiHVDNwvqI0hIFlzV8dgIr35kV1kRU= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0/go.mod h1:NfchwuyNoMcZ5MLHwPrODwUF1HWCXWrL31s8gSAdIKY= -go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8= -go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM= +go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= +go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 h1:GqRJVj7UmLjCVyVJ3ZFLdPRmhDUp2zFmQe3RHIOsw24= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0/go.mod h1:ri3aaHSmCTVYu2AWv44YMauwAQc0aqI9gHKIcSbI1pU= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 h1:aTL7F04bJHUlztTsNGJ2l+6he8c+y/b//eR0jjjemT4= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0/go.mod h1:kldtb7jDTeol0l3ewcmd8SDvx3EmIE7lyvqbasU3QC4= go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.38.0 h1:kJxSDN4SgWWTjG/hPp3O7LCGLcHXFlvS2/FFOrwL+SE= go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.38.0/go.mod h1:mgIOzS7iZeKJdeB8/NYHrJ48fdGc71Llo5bJ1J4DWUE= -go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA= -go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI= -go.opentelemetry.io/otel/sdk v1.38.0 h1:l48sr5YbNf2hpCUj/FoGhW9yDkl+Ma+LrVl8qaM5b+E= -go.opentelemetry.io/otel/sdk v1.38.0/go.mod h1:ghmNdGlVemJI3+ZB5iDEuk4bWA3GkTpW+DOoZMYBVVg= -go.opentelemetry.io/otel/sdk/metric v1.38.0 h1:aSH66iL0aZqo//xXzQLYozmWrXxyFkBJ6qT5wthqPoM= -go.opentelemetry.io/otel/sdk/metric v1.38.0/go.mod h1:dg9PBnW9XdQ1Hd6ZnRz689CbtrUp0wMMs9iPcgT9EZA= -go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE= -go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs= +go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0= +go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs= +go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= +go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= +go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= +go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= +go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= +go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= go.opentelemetry.io/proto/otlp v1.7.1 h1:gTOMpGDb0WTBOP8JaO72iL3auEZhVmAQg4ipjOVAtj4= go.opentelemetry.io/proto/otlp v1.7.1/go.mod h1:b2rVh6rfI/s2pHWNlB7ILJcRALpcNDzKhACevjI+ZnE= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= @@ -773,8 +773,8 @@ golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qx golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= -golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= -golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= +golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190523182746-aaccbc9213b0/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -1011,8 +1011,8 @@ google.golang.org/grpc v1.28.0/go.mod h1:rpkK4SK4GF4Ach/+MFLZUBavHOvF2JJB5uozKKa google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk= google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak= google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak= -google.golang.org/grpc v1.75.0 h1:+TW+dqTd2Biwe6KKfhE5JpiYIBWq865PhKGSXiivqt4= -google.golang.org/grpc v1.75.0/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ= +google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= +google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= diff --git a/pkg/db/context.go b/pkg/db/context.go index e1ffef7f..63561e93 100755 --- a/pkg/db/context.go +++ b/pkg/db/context.go @@ -2,6 +2,7 @@ package db import ( "context" + "errors" "github.com/google/uuid" "github.com/prometheus/client_golang/prometheus" @@ -88,12 +89,26 @@ func MarkForRollback(ctx context.Context, err error) { // NewAdvisoryLockContext returns a new context with AdvisoryLock stored in it. // Upon error, the original context is still returned along with an error. // +// IMPORTANT: Advisory locks are for cross-pod coordination (e.g., migrations, scheduled jobs), +// NOT for database row-level concurrency. For row-level concurrency, use SELECT FOR UPDATE. +// // CONCURRENCY: The returned context must not be shared across goroutines that call // NewAdvisoryLockContext or Unlock concurrently, as the internal lock map is not // protected by a mutex. Each goroutine should derive its own context chain. func NewAdvisoryLockContext( ctx context.Context, connection SessionFactory, id string, lockType LockType, ) (context.Context, string, error) { + // FAIL-FAST: Detect transaction created before advisory lock + if _, hasTransaction := dbContext.Transaction(ctx); hasTransaction { + return ctx, "", errors.New( + "advisory lock cannot be acquired within an existing transaction.\n" + + "This causes a race condition where lock is released before transaction commits.\n\n" + + "Correct patterns:\n" + + " 1. For pod coordination: Acquire lock BEFORE transaction\n" + + " 2. For row-level concurrency: Use SELECT FOR UPDATE instead", + ) + } + // lockOwnerID will be different for every service function that attempts to start a lock. // only the initial call in the stack must unlock. // Unlock() will compare UUIDs and ensure only the top level call succeeds. diff --git a/test/integration/advisory_locks_test.go b/test/integration/advisory_locks_test.go index a62221ce..dda26594 100644 --- a/test/integration/advisory_locks_test.go +++ b/test/integration/advisory_locks_test.go @@ -77,10 +77,10 @@ func acquireLock(h *test.Helper, waiter *sync.WaitGroup) { Expect(err).NotTo(HaveOccurred(), "Failed to update counter") } -// TestAdvisoryLocksWithTransactions validates that advisory locks work correctly -// when combined with database transactions in various orders. Uses actual database -// operations to prove serialization. -func TestAdvisoryLocksWithTransactions(t *testing.T) { +// TestRowLevelConcurrency validates the correct pattern for concurrent row updates +// using SELECT FOR UPDATE. This is the proper way to handle row-level concurrency +// in PostgreSQL, not advisory locks. +func TestRowLevelConcurrency(t *testing.T) { h, _ := test.RegisterIntegration(t) // Create a counter table and initialize to 0 @@ -92,92 +92,76 @@ func TestAdvisoryLocksWithTransactions(t *testing.T) { Expect(err).NotTo(HaveOccurred(), "Failed to initialize counter") defer g2.Exec("DROP TABLE IF EXISTS lock_test_counter_tx") - // Test all three transaction ordering scenarios deterministically - testCases := []struct { - name string - txBeforeLock bool - txAfterLock bool - }{ - { - name: "tx_before_lock", - txBeforeLock: true, - txAfterLock: false, - }, - { - name: "tx_after_lock", - txBeforeLock: false, - txAfterLock: true, - }, - { - name: "no_tx", - txBeforeLock: false, - txAfterLock: false, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - // Run multiple goroutines for each scenario to test concurrency - goroutines := 3 - var waiter sync.WaitGroup - waiter.Add(goroutines) - - for i := 0; i < goroutines; i++ { - go acquireLockWithTransaction(h, &waiter, tc.txBeforeLock, tc.txAfterLock) - } + total := 10 + var waiter sync.WaitGroup + waiter.Add(total) - waiter.Wait() - }) + // Simulate concurrent updates using SELECT FOR UPDATE (correct pattern) + for i := 0; i < total; i++ { + go incrementCounterWithRowLock(h, &waiter) } - // All test cases combined should have incremented the counter by 9 (3 scenarios × 3 goroutines) - expectedTotal := 9 + waiter.Wait() + + // All goroutines should have incremented the counter by 1, resulting in 10 var finalValue int err = g2.Raw("SELECT value FROM lock_test_counter_tx WHERE id = 1").Scan(&finalValue).Error Expect(err).NotTo(HaveOccurred(), "Failed to read final counter value") - Expect(finalValue).To(Equal(expectedTotal), "Counter should equal total") + Expect(finalValue).To(Equal(total), "Counter should equal total (no lost updates)") } -func acquireLockWithTransaction(h *test.Helper, waiter *sync.WaitGroup, txBeforeLock bool, txAfterLock bool) { +func incrementCounterWithRowLock(h *test.Helper, waiter *sync.WaitGroup) { defer waiter.Done() ctx := context.Background() - - var dberr error - - // Add Tx before lock if requested - if txBeforeLock { - ctx, dberr = db.NewContext(ctx, h.DBFactory) - Expect(dberr).NotTo(HaveOccurred(), "Failed to create transaction context") - defer db.Resolve(ctx) - } - - // Acquire advisory lock - ctx, lockOwnerID, dberr := db.NewAdvisoryLockContext(ctx, h.DBFactory, "test-resource-tx", db.Migrations) - Expect(dberr).NotTo(HaveOccurred(), "Failed to acquire lock") - defer db.Unlock(ctx, lockOwnerID) - - // Add Tx after lock if requested - if txAfterLock { - ctx, dberr = db.NewContext(ctx, h.DBFactory) - Expect(dberr).NotTo(HaveOccurred(), "Failed to create transaction context") - defer db.Resolve(ctx) - } - g2 := h.DBFactory.New(ctx) - // Read current value from database + // Begin transaction + tx := g2.Begin() + Expect(tx.Error).NotTo(HaveOccurred(), "Failed to begin transaction") + defer tx.Rollback() + + // SELECT FOR UPDATE acquires row-level lock + // Other transactions trying to SELECT FOR UPDATE on this row will BLOCK here var currentValue int - err := g2.Raw("SELECT value FROM lock_test_counter_tx WHERE id = 1").Scan(¤tValue).Error - Expect(err).NotTo(HaveOccurred(), "Failed to read counter") + err := tx.Raw("SELECT value FROM lock_test_counter_tx WHERE id = 1 FOR UPDATE").Scan(¤tValue).Error + Expect(err).NotTo(HaveOccurred(), "Failed to read counter with row lock") - // Some slow work + // Some slow work to increase likelihood of contention time.Sleep(20 * time.Millisecond) - // Increment and save to database + // Increment and update newValue := currentValue + 1 - err = g2.Exec("UPDATE lock_test_counter_tx SET value = ? WHERE id = 1", newValue).Error + err = tx.Exec("UPDATE lock_test_counter_tx SET value = ? WHERE id = 1", newValue).Error Expect(err).NotTo(HaveOccurred(), "Failed to update counter") + + // Commit releases the row lock + err = tx.Commit().Error + Expect(err).NotTo(HaveOccurred(), "Failed to commit transaction") +} + +// TestAdvisoryLockFailsWithExistingTransaction validates that advisory locks +// properly reject attempts to acquire a lock when a transaction already exists +// in the context. This prevents the race condition where the lock is released +// before the transaction commits. +func TestAdvisoryLockFailsWithExistingTransaction(t *testing.T) { + h, _ := test.RegisterIntegration(t) + + ctx := context.Background() + + // Create transaction first + ctx, err := db.NewContext(ctx, h.DBFactory) + Expect(err).NotTo(HaveOccurred(), "Failed to create transaction context") + defer db.Resolve(ctx) + + // Try to acquire advisory lock (should fail) + _, lockID, err := db.NewAdvisoryLockContext(ctx, h.DBFactory, "test-resource", db.Migrations) + + // Should fail with clear error + Expect(err).To(HaveOccurred(), "Advisory lock should fail when transaction exists") + Expect(err.Error()).To(ContainSubstring("transaction"), "Error should mention transaction") + Expect(err.Error()).To(ContainSubstring("SELECT FOR UPDATE"), "Error should suggest SELECT FOR UPDATE") + Expect(lockID).To(BeEmpty(), "Lock ID should be empty on error") } // TestLocksAndExpectedWaits validates the behavior of advisory locks: