Method Commit

storage/storage.go:173–230

Commit implements commit method of two-phase commit worker.

(ctx context.Context, wb twopc.WriteBatch)

// Commit implements commit method of two-phase commit worker.
func (s *Storage) Commit(ctx context.Context, wb twopc.WriteBatch) (result interface{}, err error) {
	el, ok := wb.(*ExecLog)

	if !ok {
		err = errors.New("unexpected WriteBatch type")
		return
	}

	s.Lock()
	defer s.Unlock()

	if s.tx != nil {
		if equalTxID(&s.id, &TxID{el.ConnectionID, el.SeqNo, el.Timestamp}) {
			// get last insert id and affected rows result
			execResult := ExecResult{}

			for _, q := range s.queries {
				// convert arguments types
				args := make([]interface{}, len(q.Args))

				for i, v := range q.Args {
					args[i] = v
				}

				var res sql.Result
				res, err = s.tx.ExecContext(ctx, q.Pattern, args...)

				if err != nil {
					log.WithError(err).Debug("commit query failed")
					s.tx.Rollback()
					s.tx = nil
					s.queries = nil
					return
				}

				lastInsertID, _ := res.LastInsertId()
				rowsAffected, _ := res.RowsAffected()

				execResult.LastInsertID = lastInsertID
				execResult.RowsAffected += rowsAffected
			}

			s.tx.Commit()
			s.tx = nil
			s.queries = nil
			result = execResult

			return
		}

		err = fmt.Errorf("twopc: inconsistent state, currently in tx: "+
			"conn = %d, seq = %d, time = %d", s.id.ConnectionID, s.id.SeqNo, s.id.Timestamp)
		return
	}

	err = errors.New("twopc: tx not prepared")
	return
}