From 936c482bee5f0e8720d5a237065dd56d30e4acd0 Mon Sep 17 00:00:00 2001 From: Juan Hernandez Date: Fri, 5 Jun 2026 20:36:12 +0200 Subject: [PATCH] NO-ISSUE: Add `Run` method to `TxManager` and `Tx` interfaces Add a reflection-based `Run` method that executes a task function within a database transaction, automatically handling commit, rollback, and panic recovery. The task function's first parameter must be either `context.Context` (in which case the transaction is stored in the context) or `database.Tx` (passed directly). Additional parameters of any type can be supplied via variadic args. If the last return value implements `error`, it determines whether to commit or rollback. `TxManager.Run` creates a new transaction, runs the task, and ends the transaction. `Tx.Run` executes the task within an existing transaction, reporting errors but leaving lifecycle management to the caller. Assisted-by: Cursor Signed-off-by: Juan Hernandez --- .../database/dao/generic_dao_events_test.go | 152 +++---- .../database/dao/generic_dao_lock_test.go | 279 ++++++------- internal/database/database_listener_test.go | 15 +- internal/database/database_notifier_test.go | 14 +- internal/database/database_tx.go | 9 + internal/database/database_tx_manager.go | 141 +++++++ internal/database/database_tx_manager_mock.go | 19 + internal/database/database_tx_manager_test.go | 370 ++++++++++++++++++ internal/database/database_tx_mock.go | 19 + .../servers/cluster_templates_server_test.go | 39 +- internal/servers/clusters_server_test.go | 42 +- ...pute_instance_catalog_items_server_test.go | 35 +- .../compute_instance_templates_server_test.go | 38 -- internal/servers/console_server_test.go | 8 + .../servers/public_ip_pools_server_test.go | 34 +- internal/servers/servers_suite_test.go | 3 +- 16 files changed, 788 insertions(+), 429 deletions(-) diff --git a/internal/database/dao/generic_dao_events_test.go b/internal/database/dao/generic_dao_events_test.go index 14d98df0e..4c00ed835 100644 --- a/internal/database/dao/generic_dao_events_test.go +++ b/internal/database/dao/generic_dao_events_test.go @@ -37,18 +37,6 @@ var _ = Describe("Generic DAO events", func() { tenancy *auth.MockTenancyLogic ) - // runWithTx starts a transaction, runs the given function using it, and ends the transaction when it finishes. - runWithTx := func(task func(ctx context.Context)) { - tx, err := tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - defer func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }() - taskCtx := database.TxIntoContext(ctx, tx) - task(taskCtx) - } - BeforeEach(func() { var err error @@ -87,7 +75,7 @@ var _ = Describe("Generic DAO events", func() { SetTenancyLogic(tenancy). Build() Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = tenantsDao.Create(). SetObject(&privatev1.Organization{ Id: "my-tenant", @@ -113,7 +101,7 @@ var _ = Describe("Generic DAO events", func() { Build() Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = generic.Create(). SetObject(&privatev1.Cluster{ Metadata: privatev1.Metadata_builder{ @@ -142,7 +130,7 @@ var _ = Describe("Generic DAO events", func() { Expect(err).ToNot(HaveOccurred()) var object *privatev1.Cluster - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { response, createErr := generic.Create(). SetObject(&privatev1.Cluster{ Metadata: privatev1.Metadata_builder{ @@ -157,7 +145,7 @@ var _ = Describe("Generic DAO events", func() { }) Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = generic.Update(). SetObject(&privatev1.Cluster{ Id: object.Id, @@ -189,7 +177,7 @@ var _ = Describe("Generic DAO events", func() { Expect(err).ToNot(HaveOccurred()) var object *privatev1.Cluster - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { response, createErr := generic.Create(). SetObject(&privatev1.Cluster{ Metadata: privatev1.Metadata_builder{ @@ -203,7 +191,7 @@ var _ = Describe("Generic DAO events", func() { } }) Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = generic.Delete(). SetId(object.GetId()). Do(ctx) @@ -224,19 +212,23 @@ var _ = Describe("Generic DAO events", func() { Build() Expect(err).ToNot(HaveOccurred()) var object *privatev1.Cluster - runWithTx(func(ctx context.Context) { - response, createErr := generic.Create(). - SetObject(&privatev1.Cluster{ - Metadata: privatev1.Metadata_builder{ - Tenant: "my-tenant", - }.Build(), - }). - Do(ctx) - err = createErr - if err == nil { + err = tm.Run( + ctx, + func(ctx context.Context) error { + response, err := generic.Create(). + SetObject(&privatev1.Cluster{ + Metadata: privatev1.Metadata_builder{ + Tenant: "my-tenant", + }.Build(), + }). + Do(ctx) + if err != nil { + return err + } object = response.GetObject() - } - }) + return nil + }, + ) Expect(err).To(MatchError("my error")) Expect(object).To(BeNil()) row := pool.QueryRow(ctx, "select count(*) from clusters") @@ -254,7 +246,7 @@ var _ = Describe("Generic DAO events", func() { Build() Expect(err).ToNot(HaveOccurred()) var object *privatev1.Cluster - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { response, createErr := generic.Create(). SetObject(&privatev1.Cluster{ Metadata: privatev1.Metadata_builder{ @@ -278,24 +270,32 @@ var _ = Describe("Generic DAO events", func() { }). Build() Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { - _, err = generic.Delete(). - SetId(object.GetId()). - Do(ctx) - }) + err = tm.Run( + ctx, + func(ctx context.Context) error { + _, err := generic.Delete(). + SetId(object.GetId()). + Do(ctx) + return err + }, + ) Expect(err).To(MatchError("my error")) // Check that the object is still there: var exists bool - runWithTx(func(ctx context.Context) { - response, existsErr := generic.Exists(). - SetId(object.GetId()). - Do(ctx) - err = existsErr - if err == nil { + err = tm.Run( + ctx, + func(ctx context.Context) error { + response, err := generic.Exists(). + SetId(object.GetId()). + Do(ctx) + if err != nil { + return err + } exists = response.GetExists() - } - }) + return nil + }, + ) Expect(err).ToNot(HaveOccurred()) Expect(exists).To(BeTrue()) }) @@ -308,7 +308,7 @@ var _ = Describe("Generic DAO events", func() { Build() Expect(err).ToNot(HaveOccurred()) var object *privatev1.Cluster - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { response, createErr := generic.Create(). SetObject(&privatev1.Cluster{ Metadata: privatev1.Metadata_builder{ @@ -335,23 +335,27 @@ var _ = Describe("Generic DAO events", func() { }). Build() Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { - _, err = generic.Update(). - SetObject(&privatev1.Cluster{ - Id: object.GetId(), - Metadata: privatev1.Metadata_builder{ - Tenant: "my-tenant", - }.Build(), - Status: &privatev1.ClusterStatus{ - ApiUrl: "https://your.api", - }, - }). - Do(ctx) - }) + err = tm.Run( + ctx, + func(ctx context.Context) error { + _, err := generic.Update(). + SetObject(&privatev1.Cluster{ + Id: object.GetId(), + Metadata: privatev1.Metadata_builder{ + Tenant: "my-tenant", + }.Build(), + Status: &privatev1.ClusterStatus{ + ApiUrl: "https://your.api", + }, + }). + Do(ctx) + return err + }, + ) Expect(err).To(MatchError("my error")) // Check that the object hasn't been updated: - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { getResponse, getErr := generic.Get(). SetId(object.GetId()). Do(ctx) @@ -384,7 +388,7 @@ var _ = Describe("Generic DAO events", func() { Build() Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = generic.Create(). SetObject(&privatev1.Cluster{ Metadata: privatev1.Metadata_builder{ @@ -415,15 +419,19 @@ var _ = Describe("Generic DAO events", func() { Build() Expect(err).ToNot(HaveOccurred()) - runWithTx(func(ctx context.Context) { - _, err = generic.Create(). - SetObject(&privatev1.Cluster{ - Metadata: privatev1.Metadata_builder{ - Tenant: "my-tenant", - }.Build(), - }). - Do(ctx) - }) + err = tm.Run( + ctx, + func(ctx context.Context) error { + _, err := generic.Create(). + SetObject(&privatev1.Cluster{ + Metadata: privatev1.Metadata_builder{ + Tenant: "my-tenant", + }.Build(), + }). + Do(ctx) + return err + }, + ) Expect(err).To(MatchError("my error 1")) Expect(called1).To(BeTrue()) Expect(called2).To(BeFalse()) @@ -444,7 +452,7 @@ var _ = Describe("Generic DAO events", func() { // Create an object with finalizers: var object *privatev1.Cluster - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { createResponse, err := generic.Create(). SetObject( privatev1.Cluster_builder{ @@ -462,7 +470,7 @@ var _ = Describe("Generic DAO events", func() { }) // Delete the object: - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = generic.Delete(). SetId(object.GetId()). Do(ctx) @@ -471,7 +479,7 @@ var _ = Describe("Generic DAO events", func() { // Remove the finalizers: object.Metadata.Finalizers = []string{} - runWithTx(func(ctx context.Context) { + err = tm.Run(ctx, func(ctx context.Context) { _, err = generic.Update(). SetObject(object). Do(ctx) diff --git a/internal/database/dao/generic_dao_lock_test.go b/internal/database/dao/generic_dao_lock_test.go index 55c5839c7..004eacb71 100644 --- a/internal/database/dao/generic_dao_lock_test.go +++ b/internal/database/dao/generic_dao_lock_test.go @@ -80,185 +80,155 @@ var _ = Describe("Lock", func() { Expect(err).ToNot(HaveOccurred()) }) - // createObject inserts an object directly into the database using auto-commit so that it is - // visible to all transactions. - createObject := func(id string, tenant string) { - _, err := pool.Exec( - ctx, - "insert into objects (id, tenant, data) values ($1, $2, '{}')", - id, tenant, - ) + // createObject inserts an object directly into the database using auto-commit so that it is visible to all + // transactions. + createObject := func(ctx context.Context, id string, tenant string) { + err := tm.Run(ctx, func(tx database.Tx) { + _, err := tx.Exec( + ctx, + "insert into objects (id, tenant, data) values ($1, $2, '{}')", + id, tenant, + ) + Expect(err).ToNot(HaveOccurred()) + }) Expect(err).ToNot(HaveOccurred()) } - // checkLocked verifies that the given identifiers correspond to rows that are currently locked - // by using 'for update skip locked' to detect locked rows. If a row is locked by another - // transaction, 'skip locked' will skip it, so an empty result means all the rows are locked. - checkLocked := func(ids ...string) { - rows, err := pool.Query( + // fetchUnlocked fetches the identifiers of the objects that are not locked by using 'for update skip locked'. + fetchUnlocked := func(ctx context.Context, ids ...string) []string { + var result []string + err := tm.Run( ctx, - "select id from objects where id = any($1) for update skip locked", - ids, + func(tx database.Tx) { + rows, err := tx.Query( + ctx, + "select id from objects where id = any($1) for update skip locked", + ids, + ) + Expect(err).ToNot(HaveOccurred()) + defer rows.Close() + for rows.Next() { + var id string + err = rows.Scan(&id) + Expect(err).ToNot(HaveOccurred()) + result = append(result, id) + } + err = rows.Err() + Expect(err).ToNot(HaveOccurred()) + }, ) Expect(err).ToNot(HaveOccurred()) - defer rows.Close() + return result + } - var unlocked []string - for rows.Next() { - var id string - err = rows.Scan(&id) - Expect(err).ToNot(HaveOccurred()) - unlocked = append(unlocked, id) - } - Expect(rows.Err()).ToNot(HaveOccurred()) + // checkLocked verifies that the given identifiers correspond to rows that are currently locked by using 'for + // update skip locked' to detect locked rows. If a row is locked by another transaction, 'skip locked' will skip + // it, so an empty result means all the rows are locked. + checkLocked := func(ctx context.Context, ids ...string) { + unlocked := fetchUnlocked(ctx, ids...) Expect(unlocked).To(BeEmpty()) } - // checkNotLocked verifies that the given identifiers correspond to rows that are not locked - // by using the same 'for update skip locked' technique. If a row is not locked it will be - // returned, so all the requested rows being returned means none of them are locked. - checkNotLocked := func(ids ...string) { - rows, err := pool.Query( - ctx, - "select id from objects where id = any($1) for update skip locked", - ids, - ) - Expect(err).ToNot(HaveOccurred()) - defer rows.Close() - - var unlocked []string - for rows.Next() { - var id string - err = rows.Scan(&id) - Expect(err).ToNot(HaveOccurred()) - unlocked = append(unlocked, id) - } - Expect(rows.Err()).ToNot(HaveOccurred()) + // checkNotLocked verifies that the given identifiers correspond to rows that are not locked by using the same + // 'for update skip locked' technique. If a row is not locked it will be returned, so all the requested rows + // being returned means none of them are locked. + checkNotLocked := func(ctx context.Context, ids ...string) { + unlocked := fetchUnlocked(ctx, ids...) Expect(unlocked).To(ConsistOf(ids)) } It("Locks a single object", func() { - createObject("obj1", "my_tenant") - - tx, err := tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - - _, err = generic.Lock(). - AddId("obj1"). - Do(ctx) + err := tm.Run( + ctx, + func(ctx context.Context) { + createObject(ctx, "obj1", "my_tenant") + _, err := generic.Lock().AddId("obj1").Do(ctx) + Expect(err).ToNot(HaveOccurred()) + checkLocked(ctx, "obj1") + }, + ) Expect(err).ToNot(HaveOccurred()) - - checkLocked("obj1") }) It("Locks multiple objects", func() { - createObject("obj1", "my_tenant") - createObject("obj2", "my_tenant") - createObject("obj3", "my_tenant") - - tx, err := tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - - _, err = generic.Lock(). - AddIds("obj1", "obj2", "obj3"). - Do(ctx) + err := tm.Run( + ctx, + func(ctx context.Context) { + objects := []string{"obj1", "obj2", "obj3"} + for _, object := range objects { + createObject(ctx, object, "my_tenant") + } + _, err := generic.Lock().AddIds(objects...).Do(ctx) + Expect(err).ToNot(HaveOccurred()) + checkLocked(ctx, objects...) + }, + ) Expect(err).ToNot(HaveOccurred()) - - checkLocked("obj1", "obj2", "obj3") }) It("Fails with not found error when locking non-existent object", func() { - tx, err := tm.Begin(ctx) + err := tm.Run( + ctx, + func(ctx context.Context) { + _, err := generic.Lock().AddId("does-not-exist").Do(ctx) + Expect(err).To(HaveOccurred()) + var notFoundErr *ErrNotFound + Expect(errors.As(err, ¬FoundErr)).To(BeTrue()) + }, + ) Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - - _, err = generic.Lock(). - AddId("does-not-exist"). - Do(ctx) - Expect(err).To(HaveOccurred()) - var notFoundErr *ErrNotFound - Expect(errors.As(err, ¬FoundErr)).To(BeTrue()) - Expect(notFoundErr.IDs).To(ConsistOf("does-not-exist")) }) It("Fails when one of multiple objects doesn't exist", func() { - createObject("obj1", "my_tenant") - - tx, err := tm.Begin(ctx) + err := tm.Run( + ctx, + func(ctx context.Context) { + createObject(ctx, "obj1", "my_tenant") + _, err := generic.Lock().AddIds("obj1", "does-not-exist").Do(ctx) + Expect(err).To(HaveOccurred()) + var notFoundErr *ErrNotFound + Expect(errors.As(err, ¬FoundErr)).To(BeTrue()) + }, + ) Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - - _, err = generic.Lock(). - AddId("obj1"). - AddId("does-not-exist"). - Do(ctx) - Expect(err).To(HaveOccurred()) - var notFoundErr *ErrNotFound - Expect(errors.As(err, ¬FoundErr)).To(BeTrue()) - Expect(notFoundErr.IDs).To(ConsistOf("does-not-exist")) }) It("Unlocks object when transaction is committed", func() { - createObject("obj1", "my_tenant") - - tx, err := tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - ctx = database.TxIntoContext(ctx, tx) - _, err = generic.Lock(). - AddId("obj1"). - Do(ctx) - Expect(err).ToNot(HaveOccurred()) - - err = tx.End(ctx) + err := tm.Run( + ctx, + func(ctx context.Context) { + createObject(ctx, "obj1", "my_tenant") + _, err := generic.Lock().AddId("obj1").Do(ctx) + Expect(err).ToNot(HaveOccurred()) + checkLocked(ctx, "obj1") + }, + ) Expect(err).ToNot(HaveOccurred()) - checkNotLocked("obj1") + checkNotLocked(ctx, "obj1") }) It("Unlocks object when transaction is rolled back", func() { - createObject("obj1", "my_tenant") - - tx, err := tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - ctx = database.TxIntoContext(ctx, tx) - _, err = generic.Lock(). - AddId("obj1"). - Do(ctx) - Expect(err).ToNot(HaveOccurred()) - - rollbackErr := errors.New("force rollback") - tx.ReportError(&rollbackErr) - err = tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - checkNotLocked("obj1") + err := tm.Run( + ctx, + func(ctx context.Context) error { + createObject(ctx, "obj1", "my_tenant") + _, err := generic.Lock().AddId("obj1").Do(ctx) + Expect(err).ToNot(HaveOccurred()) + return errors.New("my error") + }, + ) + Expect(err).To(MatchError("my error")) + checkNotLocked(ctx, "obj1") }) It("Prevents deadlocks by locking in consistent order", func() { - createObject("a", "my_tenant") - createObject("b", "my_tenant") + createObject(ctx, "a", "my_tenant") + createObject(ctx, "b", "my_tenant") // Start a transaction and lock 'a' using direct SQL: - tx1, err := pool.Begin(ctx) + tx, err := pool.Begin(ctx) Expect(err).ToNot(HaveOccurred()) - defer tx1.Rollback(ctx) - _, err = tx1.Exec(ctx, "select id from objects where id = 'a' for update") + _, err = tx.Exec(ctx, "select id from objects where id = 'a' for update") Expect(err).ToNot(HaveOccurred()) // Start a goroutine that tries to lock 'b' and 'a' (in that order) via the DAO. @@ -267,29 +237,26 @@ var _ = Describe("Lock", func() { done := make(chan error, 1) go func() { defer GinkgoRecover() - tx2, err := tm.Begin(ctx) - if err != nil { - done <- err - return - } - ctx := database.TxIntoContext(ctx, tx2) - _, lockErr := generic.Lock(). - AddIds("b", "a"). - Do(ctx) - err = tx2.End(ctx) - done <- errors.Join(lockErr, err) + err := tm.Run( + ctx, + func(ctx context.Context) error { + _, err := generic.Lock().AddIds("b", "a").Do(ctx) + return err + }, + ) + done <- err }() - // Give the goroutine time to reach the blocking lock on 'a' and verify that it - // doesn't complete while 'a' is held: + // Give the goroutine time to reach the blocking lock on 'a' and verify that it doesn't complete while + // 'a' is held: Consistently(done, 100*time.Millisecond).ShouldNot(Receive()) - // Verify that 'b' is not locked, proving that the DAO tried to lock 'a' first - // even though 'b' was passed first: - checkNotLocked("b") + // Verify that 'b' is not locked, proving that the DAO tried to lock 'a' first even though 'b' was + // passed first: + checkNotLocked(ctx, "b") // Release 'a' by committing, allowing the goroutine to proceed: - err = tx1.Commit(ctx) + err = tx.Commit(ctx) Expect(err).ToNot(HaveOccurred()) // Verify the goroutine completed without error: diff --git a/internal/database/database_listener_test.go b/internal/database/database_listener_test.go index 948761e86..4b5d5359d 100644 --- a/internal/database/database_listener_test.go +++ b/internal/database/database_listener_test.go @@ -159,20 +159,17 @@ var _ = Describe("Listener", func() { }) // notify sends a payload through the database notification channel. - notify := func(payload proto.Message) { - tx, err := tm.Begin(ctx) - defer func() { - err := tx.End(ctx) + notify := func(ctx context.Context, payload proto.Message) { + err := tm.Run(ctx, func(ctx context.Context) { + err := notifier.Notify(ctx, payload) Expect(err).ToNot(HaveOccurred()) - }() - ctx = TxIntoContext(ctx, tx) - err = notifier.Notify(ctx, payload) + }) Expect(err).ToNot(HaveOccurred()) } It("Receives one notification", func() { sent := wrapperspb.String("my payload") - notify(sent) + notify(ctx, sent) var received *wrapperspb.StringValue Eventually(payloads).Should(Receive(&received)) Expect(proto.Equal(received, sent)).To(BeTrue()) @@ -192,7 +189,7 @@ var _ = Describe("Listener", func() { "nueve", } for _, value := range sent { - notify(wrapperspb.String(value)) + notify(ctx, wrapperspb.String(value)) } var received []string for range len(sent) { diff --git a/internal/database/database_notifier_test.go b/internal/database/database_notifier_test.go index db6fea305..f2478df2a 100644 --- a/internal/database/database_notifier_test.go +++ b/internal/database/database_notifier_test.go @@ -54,16 +54,6 @@ var _ = Describe("Notifier", func() { Expect(err).ToNot(HaveOccurred()) }) - // runWithTx starts a transaction, runs the given function using it, and ends the transaction when it finishes. - runWithTx := func(task func(ctx context.Context)) { - tx, err := tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - taskCtx := TxIntoContext(ctx, tx) - task(taskCtx) - err = tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - } - Describe("Creation", func() { It("Can be created when all the required parameters are set", func() { notifier, err := NewNotifier(). @@ -137,9 +127,7 @@ var _ = Describe("Notifier", func() { // Send the notification: payload := wrapperspb.Int32(42) - runWithTx(func(ctx context.Context) { - err = notifier.Notify(ctx, payload) - }) + err = tm.Run(ctx, notifier.Notify, payload) Expect(err).ToNot(HaveOccurred()) }) diff --git a/internal/database/database_tx.go b/internal/database/database_tx.go index 4c1aae9b0..a805972b0 100644 --- a/internal/database/database_tx.go +++ b/internal/database/database_tx.go @@ -63,6 +63,15 @@ type Tx interface { // It this method is called multiple times for the same transaction the reported errors will be accumulated. ReportError(err *error) + // Run executes the given task function within this transaction. If the task returns an error or panics, the + // error is reported to the transaction (marking it for rollback) and returned to the caller. The transaction is + // not ended by this method; the caller is still responsible for calling End. + // + // The task must be a function whose first parameter is either context.Context or Tx. Any additional parameters + // are passed via args. If the last return value implements the error interface, it will be used to determine + // the outcome. + Run(ctx context.Context, task any, args ...any) error + // End finishes a transaction. It will be committed if no errors have been reported, or rolled back otherwise. // See the ReportError method for details on how errors are tracked. End(ctx context.Context) error diff --git a/internal/database/database_tx_manager.go b/internal/database/database_tx_manager.go index e98247d03..d2011151c 100644 --- a/internal/database/database_tx_manager.go +++ b/internal/database/database_tx_manager.go @@ -16,7 +16,9 @@ package database import ( "context" "errors" + "fmt" "log/slog" + "reflect" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" @@ -29,6 +31,19 @@ import ( type TxManager interface { // Begin starts a new transaction. Begin(ctx context.Context) (Tx, error) + + // Run starts a new transaction, executes the given task function, and then commits the transaction if the task + // completes without error and without panic. If the task returns an error, panics, or marks the transaction for + // rollback via ReportError, the transaction is rolled back instead. + // + // The task must be a function whose first parameter is either context.Context or Tx. Any additional parameters + // are passed via args. When the first parameter is context.Context, the transaction is stored in it and can be + // retrieved with TxFromContext. When the first parameter is Tx, the transaction is passed directly. + // + // If the last return value of the task implements the error interface, it will be used to determine whether to + // commit or rollback. Functions with no return values or without a trailing error are also accepted (they + // commit unless a panic or ReportError occurs). + Run(ctx context.Context, task any, args ...any) error } // TxManagerBuilder is a builder responsible for constructing database transaction managers. Don't create instances of @@ -92,6 +107,26 @@ func (m *txManager) Begin(ctx context.Context) (tx Tx, err error) { return } +// Run starts a transaction, invokes the task, and commits or rolls back depending on the outcome. +func (m *txManager) Run(ctx context.Context, task any, args ...any) error { + tx, err := m.Begin(ctx) + if err != nil { + return err + } + taskErr := runTxTask(ctx, tx, task, args) + if taskErr != nil { + tx.ReportError(&taskErr) + } + endErr := tx.End(ctx) + if taskErr != nil { + if endErr != nil { + return errors.Join(taskErr, endErr) + } + return taskErr + } + return endErr +} + func (t *managedTx) End(ctx context.Context) error { if t.real == nil { return nil @@ -151,6 +186,14 @@ func (t *managedTx) ReportError(err *error) { } } +func (t *managedTx) Run(ctx context.Context, task any, args ...any) error { + taskErr := runTxTask(ctx, t, task, args) + if taskErr != nil { + t.ReportError(&taskErr) + } + return taskErr +} + // ensureReal makes sure that the real transaction exists, creating it if needed. func (t *managedTx) ensureReal(ctx context.Context) error { if t.real != nil { @@ -172,3 +215,101 @@ type managedRow struct { func (r *managedRow) Scan(dest ...any) error { return r.err } + +// runTask validates the task function signature, builds the argument list, calls the function via reflection, and +// extracts the trailing error return value if present. +func runTxTask(ctx context.Context, tx Tx, task any, args []any) (taskErr error) { + defer func() { + if r := recover(); r != nil { + switch v := r.(type) { + case error: + taskErr = fmt.Errorf("task panicked: %w", v) + default: + taskErr = fmt.Errorf("task panicked: %v", v) + } + } + }() + + // Check that the task function is acceptable: + taskFunc := reflect.ValueOf(task) + taskType := taskFunc.Type() + if taskType.Kind() != reflect.Func { + taskErr = fmt.Errorf("task must be a function, got '%T'", task) + return + } + if taskType.NumIn() == 0 { + taskErr = errors.New("task function must have at least one parameter, context or transaction") + return + } + + // Check that the first parameter of the task is either a context or a transaction: + firstParam := taskType.In(0) + var firstArg reflect.Value + switch { + case firstParam.Implements(contextType): + firstArg = reflect.ValueOf(TxIntoContext(ctx, tx)) + case firstParam == txType: + firstArg = reflect.ValueOf(tx) + default: + taskErr = fmt.Errorf( + "first parameter of task function must be context.Context or database.Tx, got %s", firstParam, + ) + return + } + + // Check that we got exactly the number of expected arguments: + expectedArgs := taskType.NumIn() - 1 + if len(args) != expectedArgs { + taskErr = fmt.Errorf("task function expects %d additional argument(s), got %d", expectedArgs, len(args)) + return + } + + // Validate and build the argument list: + callArgs := make([]reflect.Value, 0, taskType.NumIn()) + callArgs = append(callArgs, firstArg) + for i, arg := range args { + paramType := taskType.In(i + 1) + if arg == nil { + switch paramType.Kind() { + case reflect.Pointer, reflect.Slice, reflect.Map, reflect.Func, reflect.Chan, reflect.Interface: + callArgs = append(callArgs, reflect.Zero(paramType)) + default: + taskErr = fmt.Errorf( + "argument %d is nil but parameter type %s is not nilable", + i+1, paramType, + ) + return + } + } else { + argType := reflect.TypeOf(arg) + if !argType.AssignableTo(paramType) { + taskErr = fmt.Errorf( + "argument %d has type %s which is not assignable to parameter type %s", + i+1, argType, paramType, + ) + return + } + callArgs = append(callArgs, reflect.ValueOf(arg)) + } + } + + // Call the task function: + results := taskFunc.Call(callArgs) + + // Extract the trailing error return value if present: + numOut := taskType.NumOut() + if numOut > 0 && taskType.Out(numOut-1).Implements(errorType) { + lastResult := results[numOut-1] + if !lastResult.IsNil() { + taskErr = lastResult.Interface().(error) + } + } + return +} + +// Well-known reflection types: +var ( + contextType = reflect.TypeFor[context.Context]() + txType = reflect.TypeFor[Tx]() + errorType = reflect.TypeFor[error]() +) diff --git a/internal/database/database_tx_manager_mock.go b/internal/database/database_tx_manager_mock.go index 3a9422514..b742e2603 100644 --- a/internal/database/database_tx_manager_mock.go +++ b/internal/database/database_tx_manager_mock.go @@ -54,3 +54,22 @@ func (mr *MockTxManagerMockRecorder) Begin(ctx any) *gomock.Call { mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Begin", reflect.TypeOf((*MockTxManager)(nil).Begin), ctx) } + +// Run mocks base method. +func (m *MockTxManager) Run(ctx context.Context, task any, args ...any) error { + m.ctrl.T.Helper() + varargs := []any{ctx, task} + for _, a := range args { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "Run", varargs...) + ret0, _ := ret[0].(error) + return ret0 +} + +// Run indicates an expected call of Run. +func (mr *MockTxManagerMockRecorder) Run(ctx, task any, args ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, task}, args...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockTxManager)(nil).Run), varargs...) +} diff --git a/internal/database/database_tx_manager_test.go b/internal/database/database_tx_manager_test.go index 9662214a4..255d9e548 100644 --- a/internal/database/database_tx_manager_test.go +++ b/internal/database/database_tx_manager_test.go @@ -3,6 +3,7 @@ package database import ( "context" "errors" + "fmt" "github.com/jackc/pgx/v5/pgxpool" . "github.com/onsi/ginkgo/v2" @@ -135,4 +136,373 @@ var _ = Describe("Transaction manager", func() { Expect(err).To(HaveOccurred()) }) }) + + Describe("Run", func() { + var manager TxManager + + BeforeEach(func() { + var err error + manager, err = NewTxManager(). + SetLogger(logger). + SetPool(pool). + Build() + Expect(err).ToNot(HaveOccurred()) + + _, err = pool.Exec(ctx, "create table my_table (my_column text)") + Expect(err).ToNot(HaveOccurred()) + }) + + It("Should accept 'func(context.Context)' and commit on success", func() { + err := manager.Run(ctx, func(ctx context.Context) { + tx, err := TxFromContext(ctx) + Expect(err).ToNot(HaveOccurred()) + _, err = tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "ctx_void") + Expect(err).ToNot(HaveOccurred()) + }) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("ctx_void")) + }) + + It("Should accept a 'func(context.Context) error' and commit when nil is returned", func() { + err := manager.Run(ctx, func(ctx context.Context) error { + tx, err := TxFromContext(ctx) + if err != nil { + return err + } + _, err = tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "ctx_err") + return err + }) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("ctx_err")) + }) + + It("Should accept a 'func(context.Context) error' and rollback when an error is returned", func() { + taskErr := fmt.Errorf("something went wrong") + err := manager.Run(ctx, func(ctx context.Context) error { + tx, err := TxFromContext(ctx) + if err != nil { + return err + } + _, err = tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "ctx_err_fail") + Expect(err).ToNot(HaveOccurred()) + return taskErr + }) + Expect(err).To(BeIdenticalTo(taskErr)) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).To(HaveOccurred()) + }) + + It("Should accept a 'func(Tx)' and commit on success", func() { + err := manager.Run(ctx, func(tx Tx) { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_void") + Expect(err).ToNot(HaveOccurred()) + }) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("tx_void")) + }) + + It("Should accept a 'func(Tx) error' and commit when nil is returned", func() { + err := manager.Run(ctx, func(tx Tx) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_err") + return err + }) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("tx_err")) + }) + + It("Should accept a 'func(Tx) error' and rollback when an error is returned", func() { + taskErr := fmt.Errorf("tx failed") + err := manager.Run(ctx, func(tx Tx) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_err_fail") + Expect(err).ToNot(HaveOccurred()) + return taskErr + }) + Expect(err).To(BeIdenticalTo(taskErr)) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).To(HaveOccurred()) + }) + + It("Should rollback when the task panics", func() { + err := manager.Run(ctx, func(tx Tx) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "panic_val") + Expect(err).ToNot(HaveOccurred()) + panic("unexpected failure") + }) + Expect(err).To(MatchError(ContainSubstring("task panicked"))) + Expect(err).To(MatchError(ContainSubstring("unexpected failure"))) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).To(HaveOccurred()) + }) + + It("Should rollback when 'ReportError' is called inside the task", func() { + err := manager.Run(ctx, func(ctx context.Context) { + tx, err := TxFromContext(ctx) + Expect(err).ToNot(HaveOccurred()) + _, err = tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "report_err") + Expect(err).ToNot(HaveOccurred()) + reportedErr := fmt.Errorf("reported error") + tx.ReportError(&reportedErr) + }) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).To(HaveOccurred()) + }) + + It("Should pass extra arguments to the task function", func() { + err := manager.Run( + ctx, + func(tx Tx, table string, value string) error { + _, err := tx.Exec( + ctx, + fmt.Sprintf("insert into %s (my_column) values ($1)", table), + value, + ) + return err + }, + "my_table", + "extra_args", + ) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("extra_args")) + }) + + It("Should accept a task with multiple return values where the last is error", func() { + err := manager.Run(ctx, func(tx Tx, val string) (int, error) { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", val) + return 42, err + }, "multi_return") + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("multi_return")) + }) + + It("Should accept a task with return values where the last is not error", func() { + err := manager.Run(ctx, func(tx Tx) int { + _, txErr := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "no_err_return") + Expect(txErr).ToNot(HaveOccurred()) + return 7 + }) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("no_err_return")) + }) + + It("Should return an error if task is not a function", func() { + err := manager.Run(ctx, "not a function") + Expect(err).To(MatchError(ContainSubstring("task must be a function"))) + }) + + It("Should return an error if first parameter is not context or Tx", func() { + err := manager.Run(ctx, func(s string) {}) + Expect(err).To(MatchError(ContainSubstring("first parameter of task function must be"))) + }) + + It("Should return an error if argument count does not match", func() { + err := manager.Run(ctx, func(tx Tx, a string, b int) error { + return nil + }, "only one arg") + Expect(err).To(MatchError(ContainSubstring("expects 2 additional argument(s), got 1"))) + }) + + It("Should handle nil arguments", func() { + type config struct{ Name string } + err := manager.Run(ctx, func(tx Tx, cfg *config) error { + Expect(cfg).To(BeNil()) + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "nil_arg") + return err + }, nil) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("nil_arg")) + }) + + It("Should return an error if nil is passed for a non-nilable parameter", func() { + err := manager.Run(ctx, func(tx Tx, n int) error { + return nil + }, nil) + Expect(err).To(MatchError(ContainSubstring( + "argument 1 is nil but parameter type int is not nilable", + ))) + }) + + It("Should return an error if argument type does not match parameter type", func() { + err := manager.Run(ctx, func(tx Tx, n int) error { + return nil + }, "not an int") + Expect(err).To(MatchError(ContainSubstring( + "argument 1 has type string which is not assignable to parameter type int", + ))) + }) + }) + + Describe("Transaction run", func() { + var manager TxManager + + BeforeEach(func() { + var err error + manager, err = NewTxManager(). + SetLogger(logger). + SetPool(pool). + Build() + Expect(err).ToNot(HaveOccurred()) + + _, err = pool.Exec(ctx, "create table my_table (my_column text)") + Expect(err).ToNot(HaveOccurred()) + }) + + It("Should commit when task succeeds", func() { + tx, err := manager.Begin(ctx) + Expect(err).ToNot(HaveOccurred()) + + err = tx.Run(ctx, func(tx Tx) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_run_ok") + return err + }) + Expect(err).ToNot(HaveOccurred()) + + err = tx.End(ctx) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("tx_run_ok")) + }) + + It("Should report the error and rollback when task returns an error", func() { + tx, err := manager.Begin(ctx) + Expect(err).ToNot(HaveOccurred()) + + taskErr := fmt.Errorf("task failed") + err = tx.Run(ctx, func(tx Tx) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_run_err") + Expect(err).ToNot(HaveOccurred()) + return taskErr + }) + Expect(err).To(BeIdenticalTo(taskErr)) + + err = tx.End(ctx) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).To(HaveOccurred()) + }) + + It("Should report the error and rollback when task panics", func() { + tx, err := manager.Begin(ctx) + Expect(err).ToNot(HaveOccurred()) + + err = tx.Run(ctx, func(tx Tx) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_run_panic") + Expect(err).ToNot(HaveOccurred()) + panic("boom") + }) + Expect(err).To(MatchError(ContainSubstring("task panicked"))) + Expect(err).To(MatchError(ContainSubstring("boom"))) + + err = tx.End(ctx) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).To(HaveOccurred()) + }) + + It("Should pass extra arguments to the task", func() { + tx, err := manager.Begin(ctx) + Expect(err).ToNot(HaveOccurred()) + + err = tx.Run(ctx, func(tx Tx, value string) error { + _, err := tx.Exec(ctx, "insert into my_table (my_column) values ($1)", value) + return err + }, "tx_run_args") + Expect(err).ToNot(HaveOccurred()) + + err = tx.End(ctx) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("tx_run_args")) + }) + + It("Should accept a context-based task", func() { + tx, err := manager.Begin(ctx) + Expect(err).ToNot(HaveOccurred()) + + err = tx.Run(ctx, func(ctx context.Context) error { + innerTx, err := TxFromContext(ctx) + if err != nil { + return err + } + _, err = innerTx.Exec(ctx, "insert into my_table (my_column) values ($1)", "tx_run_ctx") + return err + }) + Expect(err).ToNot(HaveOccurred()) + + err = tx.End(ctx) + Expect(err).ToNot(HaveOccurred()) + + row := pool.QueryRow(ctx, "select my_column from my_table") + var value string + err = row.Scan(&value) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal("tx_run_ctx")) + }) + }) }) diff --git a/internal/database/database_tx_mock.go b/internal/database/database_tx_mock.go index af9aad442..0de4944f8 100644 --- a/internal/database/database_tx_mock.go +++ b/internal/database/database_tx_mock.go @@ -126,3 +126,22 @@ func (mr *MockTxMockRecorder) ReportError(err any) *gomock.Call { mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReportError", reflect.TypeOf((*MockTx)(nil).ReportError), err) } + +// Run mocks base method. +func (m *MockTx) Run(ctx context.Context, task any, args ...any) error { + m.ctrl.T.Helper() + varargs := []any{ctx, task} + for _, a := range args { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "Run", varargs...) + ret0, _ := ret[0].(error) + return ret0 +} + +// Run indicates an expected call of Run. +func (mr *MockTxMockRecorder) Run(ctx, task any, args ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, task}, args...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockTx)(nil).Run), varargs...) +} diff --git a/internal/servers/cluster_templates_server_test.go b/internal/servers/cluster_templates_server_test.go index 3f2d2ac5f..b1831038d 100644 --- a/internal/servers/cluster_templates_server_test.go +++ b/internal/servers/cluster_templates_server_test.go @@ -14,7 +14,6 @@ language governing permissions and limitations under the License. package servers import ( - "context" "fmt" . "github.com/onsi/ginkgo/v2" @@ -26,42 +25,6 @@ import ( ) var _ = Describe("Cluster templates server", func() { - var ( - ctx context.Context - tx database.Tx - ) - - BeforeEach(func() { - var err error - - // Create a context: - ctx = context.Background() - - // Prepare the database pool: - db, err := server.NewInstance().Build() - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(db.Close) - pool, err := db.Pool(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(pool.Close) - - // Create the transaction manager: - tm, err := database.NewTxManager(). - SetLogger(logger). - SetPool(pool). - Build() - Expect(err).ToNot(HaveOccurred()) - - // Start a transaction and add it to the context: - tx, err = tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - }) - Describe("Creation", func() { It("Can be built if all the required parameters are set", func() { server, err := NewClusterTemplatesServer(). @@ -295,6 +258,8 @@ var _ = Describe("Cluster templates server", func() { // Add a finalizer, as otherwise the object will be immediatelly deleted and archived and it // won't be possible to verify the deletion timestamp. This can't be done using the server // because this is a public object, and public objects don't have the finalizers field. + tx, err := database.TxFromContext(ctx) + Expect(err).ToNot(HaveOccurred()) _, err = tx.Exec( ctx, `update cluster_templates set finalizers = '{"a"}' where id = $1`, diff --git a/internal/servers/clusters_server_test.go b/internal/servers/clusters_server_test.go index bc78fe2bb..3d22bb5be 100644 --- a/internal/servers/clusters_server_test.go +++ b/internal/servers/clusters_server_test.go @@ -14,7 +14,6 @@ language governing permissions and limitations under the License. package servers import ( - "context" "fmt" . "github.com/onsi/ginkgo/v2" @@ -29,53 +28,20 @@ import ( privatev1 "github.com/osac-project/fulfillment-service/internal/api/osac/private/v1" publicv1 "github.com/osac-project/fulfillment-service/internal/api/osac/public/v1" "github.com/osac-project/fulfillment-service/internal/auth" - "github.com/osac-project/fulfillment-service/internal/collections" "github.com/osac-project/fulfillment-service/internal/database" "github.com/osac-project/fulfillment-service/internal/database/dao" ) var _ = Describe("Clusters server", func() { - var ( - ctx context.Context - tx database.Tx - ) - BeforeEach(func() { - var err error - - // Create a context: - ctx = context.Background() + // Put a context into the subject: ctx = auth.ContextWithSubject( ctx, &auth.Subject{ - User: "system", - Tenants: collections.NewUniversalSet[string](), + User: auth.SystemTenant, + Tenants: auth.AllTenants, }, ) - - // Prepare the database pool: - db, err := server.NewInstance().Build() - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(db.Close) - pool, err := db.Pool(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(pool.Close) - - // Create the transaction manager: - tm, err := database.NewTxManager(). - SetLogger(logger). - SetPool(pool). - Build() - Expect(err).ToNot(HaveOccurred()) - - // Start a transaction and add it to the context: - tx, err = tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) }) Describe("Creation", func() { @@ -880,6 +846,8 @@ var _ = Describe("Clusters server", func() { // Add a finalizer, as otherwise the object will be immediatelly deleted and archived and it // won't be possible to verify the deletion timestamp. This can't be done using the server // because this is a public object, and public objects don't have the finalizers field. + tx, err := database.TxFromContext(ctx) + Expect(err).ToNot(HaveOccurred()) _, err = tx.Exec( ctx, `update clusters set finalizers = '{"a"}' where id = $1`, diff --git a/internal/servers/compute_instance_catalog_items_server_test.go b/internal/servers/compute_instance_catalog_items_server_test.go index e68078683..80d20834f 100644 --- a/internal/servers/compute_instance_catalog_items_server_test.go +++ b/internal/servers/compute_instance_catalog_items_server_test.go @@ -14,7 +14,6 @@ language governing permissions and limitations under the License. package servers import ( - "context" "fmt" . "github.com/onsi/ginkgo/v2" @@ -29,38 +28,6 @@ import ( ) var _ = Describe("Compute instance catalog items server", func() { - var ( - ctx context.Context - tx database.Tx - ) - - BeforeEach(func() { - var err error - - ctx = context.Background() - - db, err := server.NewInstance().Build() - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(db.Close) - pool, err := db.Pool(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(pool.Close) - - tm, err := database.NewTxManager(). - SetLogger(logger). - SetPool(pool). - Build() - Expect(err).ToNot(HaveOccurred()) - - tx, err = tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - }) - Describe("Creation", func() { It("Can be built if all the required parameters are set", func() { server, err := NewComputeInstanceCatalogItemsServer(). @@ -355,6 +322,8 @@ var _ = Describe("Compute instance catalog items server", func() { Expect(err).ToNot(HaveOccurred()) object := createResponse.GetObject() + tx, err := database.TxFromContext(ctx) + Expect(err).ToNot(HaveOccurred()) _, err = tx.Exec( ctx, `update compute_instance_catalog_items set finalizers = '{"a"}' where id = $1`, diff --git a/internal/servers/compute_instance_templates_server_test.go b/internal/servers/compute_instance_templates_server_test.go index b74bc3a82..96b088a1e 100644 --- a/internal/servers/compute_instance_templates_server_test.go +++ b/internal/servers/compute_instance_templates_server_test.go @@ -14,7 +14,6 @@ language governing permissions and limitations under the License. package servers import ( - "context" "fmt" . "github.com/onsi/ginkgo/v2" @@ -22,46 +21,9 @@ import ( "google.golang.org/protobuf/types/known/fieldmaskpb" publicv1 "github.com/osac-project/fulfillment-service/internal/api/osac/public/v1" - "github.com/osac-project/fulfillment-service/internal/database" ) var _ = Describe("Compute instance templates server", func() { - var ( - ctx context.Context - tx database.Tx - ) - - BeforeEach(func() { - var err error - - // Create a context: - ctx = context.Background() - - // Prepare the database pool: - db, err := server.NewInstance().Build() - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(db.Close) - pool, err := db.Pool(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(pool.Close) - - // Create the transaction manager: - tm, err := database.NewTxManager(). - SetLogger(logger). - SetPool(pool). - Build() - Expect(err).ToNot(HaveOccurred()) - - // Start a transaction and add it to the context: - tx, err = tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - }) - Describe("Builder", func() { It("Creates server with logger and tenancy logic", func() { // Create the public server: diff --git a/internal/servers/console_server_test.go b/internal/servers/console_server_test.go index 07265e5b5..f753ef401 100644 --- a/internal/servers/console_server_test.go +++ b/internal/servers/console_server_test.go @@ -139,6 +139,10 @@ func (m *mockTxManager) End(ctx context.Context, tx database.Tx) error { return nil } +func (m *mockTxManager) Run(ctx context.Context, task any, args ...any) error { + return nil +} + // mockTx is a no-op transaction for testing. type mockTx struct{} @@ -158,6 +162,10 @@ func (m *mockTx) ReportError(err *error) {} func (m *mockTx) End(ctx context.Context) error { return nil } +func (m *mockTx) Run(ctx context.Context, task any, args ...any) error { + return nil +} + // newFakeHubClientFactory returns a HubClientFactory that ignores the kubeconfig // and always returns the provided fake client. func newFakeHubClientFactory(client clnt.Client) HubClientFactory { diff --git a/internal/servers/public_ip_pools_server_test.go b/internal/servers/public_ip_pools_server_test.go index 463a9ec14..272761dbe 100644 --- a/internal/servers/public_ip_pools_server_test.go +++ b/internal/servers/public_ip_pools_server_test.go @@ -14,7 +14,6 @@ language governing permissions and limitations under the License. package servers import ( - "context" "fmt" . "github.com/onsi/ginkgo/v2" @@ -25,45 +24,14 @@ import ( privatev1 "github.com/osac-project/fulfillment-service/internal/api/osac/private/v1" publicv1 "github.com/osac-project/fulfillment-service/internal/api/osac/public/v1" - "github.com/osac-project/fulfillment-service/internal/database" ) var _ = Describe("Public IP pools server", func() { - var ( - ctx context.Context - tx database.Tx - privatePool *PrivatePublicIPPoolsServer - ) + var privatePool *PrivatePublicIPPoolsServer BeforeEach(func() { var err error - ctx = context.Background() - - // Prepare the database pool: - db, err := server.NewInstance().Build() - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(db.Close) - pool, err := db.Pool(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(pool.Close) - - // Create the transaction manager: - tm, err := database.NewTxManager(). - SetLogger(logger). - SetPool(pool). - Build() - Expect(err).ToNot(HaveOccurred()) - - // Start a transaction and add it to the context: - tx, err = tm.Begin(ctx) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(func() { - err := tx.End(ctx) - Expect(err).ToNot(HaveOccurred()) - }) - ctx = database.TxIntoContext(ctx, tx) - // The private server is used to seed pool data for list/get tests. privatePool, err = NewPrivatePublicIPPoolsServer(). SetLogger(logger). diff --git a/internal/servers/servers_suite_test.go b/internal/servers/servers_suite_test.go index 35e089a22..f49906fb8 100644 --- a/internal/servers/servers_suite_test.go +++ b/internal/servers/servers_suite_test.go @@ -38,6 +38,7 @@ var ( ctrl *gomock.Controller logger *slog.Logger server *database.Container + tm database.TxManager attribution *auth.MockAttributionLogic tenancy *auth.MockTenancyLogic ) @@ -106,7 +107,7 @@ var _ = BeforeEach(func() { DeferCleanup(pool.Close) // Create the transaction manager: - tm, err := database.NewTxManager(). + tm, err = database.NewTxManager(). SetLogger(logger). SetPool(pool). Build()