// TransactionRunner is part of the Database interface.
349946
func (db *database) TransactionRunner() (runner jujutxn.Runner, closer SessionCloser) {
	runner = db.runner
	closer = dontCloseAnything
	if runner == nil {
		raw := db.raw
		if !db.ownSession {
			session := raw.Session.Copy()
			raw = raw.With(session)
			closer = session.Close
		}
		observer := func(t jujutxn.Transaction) {
			txnLogger.Tracef("ran transaction in %.3fs (retries: %d) %# v\nerr: %v",
				t.Duration.Seconds(), t.Attempt, pretty.Formatter(t.Ops), t.Error)
		}
		if db.runTransactionObserver != nil {
			observer = func(t jujutxn.Transaction) {
				txnLogger.Tracef("ran transaction in %.3fs (retries: %d) %# v\nerr: %v",
					t.Duration.Seconds(), t.Attempt, pretty.Formatter(t.Ops), t.Error)
				db.runTransactionObserver(
					db.raw.Name, db.modelUUID,
					t.Ops, t.Error,
				)
			}
		}
		params := jujutxn.RunnerParams{
			Database:               raw,
			RunTransactionObserver: observer,
			Clock:                  db.clock,
			ServerSideTransactions: db.serverSideTransactions,
		}
		runner = jujutxn.NewRunner(params)
	}
	return &multiModelRunner{
		rawRunner: runner,
		modelUUID: db.modelUUID,
		schema:    db.schema,
	}, closer
}
1215322
func (t *Transaction) Commit() error {
	t.dbmap.trace("commit;")
	return t.Tx.Commit()
}
959341
func (t *Transaction) Rollback() error {
	t.dbmap.trace("rollback;")
	return t.Tx.Rollback()
}
987268
func (db *database) RunTransactionFor(modelUUID string, ops []txn.Op) error {
	newDB, dbcloser := db.CopyForModel(modelUUID)
	defer dbcloser()
	runner, closer := newDB.TransactionRunner()
	defer closer()
	return runner.RunTransaction(&jujutxn.Transaction{Ops: ops})
}
233483
func (db *database) RunRawTransaction(ops []txn.Op) error {
	runner, closer := db.TransactionRunner()
	defer closer()
	if multiRunner, ok := runner.(*multiModelRunner); ok {
		runner = multiRunner.rawRunner
	}
	return runner.RunTransaction(&jujutxn.Transaction{Ops: ops})
}
258111
func (pgb *ChainDB) TransactionBlock(txID string) (string, uint32, int8, error) {
	ctx, cancel := context.WithTimeout(pgb.ctx, pgb.queryTimeout)
	defer cancel()
	_, blockHash, blockInd, tree, err := RetrieveTxByHash(ctx, pgb.db, txID)
	return blockHash, blockInd, tree, pgb.replaceCancelError(err)
}
165116
func RunInTransaction(c context.Context, f func(tc context.Context) error,
	opts *datastore.TransactionOptions) error {

	return datastore.RunInTransaction(c, func(tc context.Context) error {
		tx := &transaction{}
		tc = context.WithValue(tc, &transactionKey, tx)
		if err := f(tc); err != nil {
			return err
		}

		// tx.Unlock() is not called as the tx context should never be called
		//again so we rather block than allow people to misuse the context.
		tx.Lock()
		memcacheCtx, err := memcacheContext(tc)
		if err != nil {
			return err
		}
		return memcacheSetMulti(memcacheCtx, tx.lockMemcacheItems)
	}, opts)
}
213364
func (m *environMongo) RunTransaction(buildTxn jujutxn.TransactionSource) error {
	return m.state.db().Run(buildTxn)
}
643937
func (c *TxnCollector) AfterRunTransaction(dbName, modelUUID string, ops []txn.Op, err error) {
	for _, op := range ops {
		c.updateMetrics(dbName, op, err)
	}
}
1119964
func (api *DatabaseAPI) SleepUntilDatabaseRunning(id int64, timeout time.Duration, maxRetry int) error {
	handler := waitingForUpFunc(func() (hasUpDown, error) {
		return api.Read(id)
	}, maxRetry)
	return blockingPoll(handler, timeout)
}
1338397
func (txc *TxConn) Rollback(ctx context.Context, session *SafeSession) error {
	if !session.InTransaction() {
		return nil
	}
	defer session.Reset()

	return txc.runSessions(session.ShardSessions, func(s *vtgatepb.Session_ShardSession) error {
		return txc.gateway.Rollback(ctx, s.Target, s.TransactionId)
	})
}
355549
func (tc *TransCache) CommitTransaction(transID string) {
	tc.transactionMux.Lock()
	tc.transBufMux.Lock()
	tc.cacheMux.Lock() // apply all transactioned items in one shot
	for _, item := range tc.transactionBuffer[transID] {
		switch item.verb {
		case AddItem:
			tc.Set(item.cacheID, item.itemID, item.value, item.groupIDs, true, transID)
		case RemoveItem:
			tc.Remove(item.cacheID, item.itemID, true, transID)
		case RemoveGroup:
			if len(item.groupIDs) >= 1 {
				tc.RemoveGroup(item.cacheID, item.groupIDs[0], true, transID)
			}
		}
	}
	tc.cacheMux.Unlock()
	delete(tc.transactionBuffer, transID)
	tc.transBufMux.Unlock()
	tc.transactionMux.Unlock()
}
694473
func (db *DB) Transactional(f func(*Tx) error) (err error) {
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	defer func() {
		if p := recover(); p != nil {
			tx.Rollback()
			panic(p)
		} else if err != nil {
			if err2 := tx.Rollback(); err2 != nil {
				if err2 == sql.ErrTxDone {
					return
				}
				err = Errors{err, err2}
			}
		} else {
			if err = tx.Commit(); err == sql.ErrTxDone {
				err = nil
			}
		}
	}()

	err = f(tx)

	return err
}
864061
func RunInTransaction(db *sql.DB, f func(*sql.Tx) error) error {
	tx, err := db.Begin()
	if err != nil {
		return &TransactionError{nil, err}
	}
	err = f(tx)
	var txErr error
	if err == nil {
		txErr = tx.Commit()
	} else {
		txErr = tx.Rollback()
	}
	if err != nil || txErr != nil {
		return &TransactionError{err, txErr}
	}
	return nil
}
583230
func (m *Migrator) runMigration(tx *sql.Tx, dir MigrationDirection, migration Migration) error {
	shouldMigrate, err := m.shouldMigrate(tx, migration.ID, dir)
	if err != nil {
		return err
	}

	if !shouldMigrate {
		return nil
	}

	var migrate func(tx *sql.Tx) error
	switch dir {
	case Up:
		migrate = migration.Up
	default:
		migrate = migration.Down
	}

	if err := migrate(tx); err != nil {
		return &MigrationError{Migration: migration, Err: err}
	}

	var query string
	switch dir {
	case Up:
		// Yes. This is a sql injection vulnerability. This gets around
		// the different bindings for sqlite3/postgres.
		//
		// If you're running migrations from user input, you're doing
		// something wrong.
		query = fmt.Sprintf("INSERT INTO %s (version) VALUES (%d)", m.table(), migration.ID)
	default:
		query = fmt.Sprintf("DELETE FROM %s WHERE version = %d", m.table(), migration.ID)
	}

	_, err = tx.Exec(query)
	return err
}
1352376
func (c *Connection) Transaction(fn func(tx *Connection) error) error {
	return c.Dialect.Lock(func() error {
		var dberr error
		cn, err := c.NewTransaction()
		if err != nil {
			return err
		}
		err = fn(cn)
		if err != nil {
			dberr = cn.TX.Rollback()
		} else {
			dberr = cn.TX.Commit()
		}
		if err != nil {
			return errors.WithStack(err)
		}
		return errors.Wrap(dberr, "error committing or rolling back transaction")
	})

}
622555
func (runner *runner) FlushChain(table Table, chain Chain) error {
	fullArgs := makeFullArgs(table, chain)

	runner.mu.Lock()
	defer runner.mu.Unlock()

	out, err := runner.run(opFlushChain, fullArgs)
	if err != nil {
		return fmt.Errorf("error flushing chain %q: %v: %s", chain, err, out)
	}
	return nil
}