internal/database: Transact supports any isolation level

- Add an arg to Transact for the isolation level
- Remove TransactSerializable

This makes it possible to use other levels, and makes it easier to see
which level is being used for each transaction.

Change-Id: Iba5e2920b4139e5e2f0f8c6b331a658d7c84f60f
Reviewed-on: https://team-review.git.corp.google.com/c/golang/discovery/+/758942
CI-Result: Cloud Build <devtools-proctor-result-processor@system.gserviceaccount.com>
Reviewed-by: Julie Qiu <julieqiu@google.com>

4 files changed, 9 insertions, 12 deletions

diff --git a/internal/postgres/insert_module.go b/internal/postgres/insert_module.go
index 4c3e33f9..3f969f9c 100644
--- a/internal/postgres/insert_module.go
+++ b/internal/postgres/insert_module.go
@@ -70,14 +70,11 @@ func (db *DB) saveModule(ctx context.Context, m *internal.Module) (err error) {
 	ctx, span := trace.StartSpan(ctx, "saveModule")
 	defer span.End()
 
-	var transact func(context.Context, func(*database.DB) error) error
+	iso := sql.LevelDefault
 	if experiment.IsActive(ctx, internal.ExperimentInsertSerializable) {
-		transact = db.db.TransactSerializable
-	} else {
-		transact = db.db.Transact
+		iso = sql.LevelSerializable
 	}
-
-	return transact(ctx, func(tx *database.DB) error {
+	return db.db.Transact(ctx, iso, func(tx *database.DB) error {
 		moduleID, err := insertModule(ctx, tx, m)
 		if err != nil {
 			return err
@@ -645,7 +642,7 @@ func removeNonDistributableData(m *internal.Module) {
 // DeleteModule deletes a Version from the database.
 func (db *DB) DeleteModule(ctx context.Context, modulePath, version string) (err error) {
 	defer derrors.Wrap(&err, "DeleteModule(ctx, db, %q, %q)", modulePath, version)
-	return db.db.Transact(ctx, func(tx *database.DB) error {
+	return db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		// We only need to delete from the modules table. Thanks to ON DELETE
 		// CASCADE constraints, that will trigger deletions from all other tables.
 		const stmt = `DELETE FROM modules WHERE module_path=$1 AND version=$2`
diff --git a/internal/postgres/search.go b/internal/postgres/search.go
index 20cff0fe..1b697841 100644
--- a/internal/postgres/search.go
+++ b/internal/postgres/search.go
@@ -640,7 +640,7 @@ func (db *DB) UpdateSearchDocumentsImportedByCount(ctx context.Context) (nUpdate
 	if err != nil {
 		return 0, err
 	}
-	err = db.db.Transact(ctx, func(tx *database.DB) error {
+	err = db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		if err := insertImportedByCounts(ctx, tx, counts); err != nil {
 			return err
 		}
@@ -905,7 +905,7 @@ func isInternalPackage(path string) bool {
 func (db *DB) DeleteOlderVersionFromSearchDocuments(ctx context.Context, modulePath, version string) (err error) {
 	defer derrors.Wrap(&err, "DeleteOlderVersionFromSearchDocuments(ctx, %q, %q)", modulePath, version)
 
-	return db.db.Transact(ctx, func(tx *database.DB) error {
+	return db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		// Collect all package paths in search_documents with the given module path
 		// and an older version. (package_path is the primary key of search_documents.)
 		var ppaths []string
diff --git a/internal/postgres/test_helper.go b/internal/postgres/test_helper.go
index efe8bbd8..9d920b02 100644
--- a/internal/postgres/test_helper.go
+++ b/internal/postgres/test_helper.go
@@ -102,7 +102,7 @@ func SetupTestDB(dbName string) (_ *DB, err error) {
 func ResetTestDB(db *DB, t *testing.T) {
 	ctx := context.Background()
 	t.Helper()
-	if err := db.db.Transact(ctx, func(tx *database.DB) error {
+	if err := db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		if _, err := tx.Exec(ctx, `
 			TRUNCATE modules CASCADE;
 			TRUNCATE version_map;
diff --git a/internal/postgres/versionstate.go b/internal/postgres/versionstate.go
index 22a38e13..f2bfc99d 100644
--- a/internal/postgres/versionstate.go
+++ b/internal/postgres/versionstate.go
@@ -36,7 +36,7 @@ func (db *DB) InsertIndexVersions(ctx context.Context, versions []*internal.Inde
 		DO UPDATE SET
 			index_timestamp=excluded.index_timestamp,
 			next_processed_after=CURRENT_TIMESTAMP`
-	return db.db.Transact(ctx, func(tx *database.DB) error {
+	return db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		return tx.BulkInsert(ctx, "module_version_states", cols, vals, conflictAction)
 	})
 }
@@ -57,7 +57,7 @@ func (db *DB) UpsertModuleVersionState(ctx context.Context, modulePath, vers, ap
 		numPackages = &n
 	}
 
-	return db.db.Transact(ctx, func(tx *database.DB) error {
+	return db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		if err := upsertModuleVersionState(ctx, tx, modulePath, vers, appVersion, numPackages, timestamp, status, goModPath, fetchErr); err != nil {
 			return err
 		}