Update WithRetryableTransaction method

Change the method signature to accept `sql.TxOptions` as the second
argument. This allows transactions to run at isolation levels other than
the default `READ COMMITTED` level.

pkg/backfill/backfill.go

@@ -169,7 +169,7 @@ type batcher struct {
 }
 
 func (b *batcher) updateBatch(ctx context.Context, conn db.DB) error {
-	return conn.WithRetryableTransaction(ctx, func(ctx context.Context, tx *sql.Tx) error {
+	return conn.WithRetryableTransaction(ctx, nil, func(ctx context.Context, tx *sql.Tx) error {
 		// Build the query to update the next batch of rows
 		sql, err := templates.BuildSQL(b.BatchConfig)
 		if err != nil {

pkg/db/db.go

@@ -21,7 +21,7 @@ const (
 type DB interface {
 	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
 	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
-	WithRetryableTransaction(ctx context.Context, f func(context.Context, *sql.Tx) error) error
+	WithRetryableTransaction(ctx context.Context, opts *sql.TxOptions, f func(context.Context, *sql.Tx) error) error
 	RawConn() *sql.DB
 	Close() error
 }
@@ -77,11 +77,11 @@ func (db *RDB) QueryContext(ctx context.Context, query string, args ...interface
 }
 
 // WithRetryableTransaction runs `f` in a transaction, retrying on lock_timeout errors.
-func (db *RDB) WithRetryableTransaction(ctx context.Context, f func(context.Context, *sql.Tx) error) error {
+func (db *RDB) WithRetryableTransaction(ctx context.Context, opts *sql.TxOptions, f func(context.Context, *sql.Tx) error) error {
 	b := backoff.New(maxBackoffDuration, backoffInterval)
 
 	for {
-		tx, err := db.DB.BeginTx(ctx, nil)
+		tx, err := db.DB.BeginTx(ctx, opts)
 		if err != nil {
 			return err
 		}

pkg/db/db_test.go

@@ -123,7 +123,7 @@ func TestWithRetryableTransaction(t *testing.T) {
 
 		// run a transaction that should retry until the lock is released
 		rdb := &db.RDB{DB: conn}
-		err := rdb.WithRetryableTransaction(ctx, func(ctx context.Context, tx *sql.Tx) error {
+		err := rdb.WithRetryableTransaction(ctx, nil, func(ctx context.Context, tx *sql.Tx) error {
 			return tx.QueryRowContext(ctx, "SELECT 1 FROM test").Err()
 		})
 		require.NoError(t, err)
@@ -149,7 +149,7 @@ func TestWithRetryableTransactionWhenContextCancelled(t *testing.T) {
 		// Cancel the context before the lock times out
 		go time.AfterFunc(500*time.Millisecond, cancel)
 
-		err := rdb.WithRetryableTransaction(ctx, func(ctx context.Context, tx *sql.Tx) error {
+		err := rdb.WithRetryableTransaction(ctx, nil, func(ctx context.Context, tx *sql.Tx) error {
 			return tx.QueryRowContext(ctx, "SELECT 1 FROM test").Err()
 		})
 		require.Errorf(t, err, "context canceled")

pkg/db/fake.go

@@ -19,7 +19,7 @@ func (db *FakeDB) QueryContext(ctx context.Context, query string, args ...interf
 	return nil, nil
 }
 
-func (db *FakeDB) WithRetryableTransaction(ctx context.Context, f func(context.Context, *sql.Tx) error) error {
+func (db *FakeDB) WithRetryableTransaction(ctx context.Context, opts *sql.TxOptions, f func(context.Context, *sql.Tx) error) error {
 	return nil
 }