Distributed transaction coordination. More...

#include <async_operations.h>

Collaboration diagram for database::async::transaction_coordinator:

Classes
struct	distributed_transaction

Public Types
enum class	transaction_state { active , preparing , prepared , committing , committed , aborting , aborted }

Public Member Functions
	transaction_coordinator ()=default
	Default constructor - used by database_context.

std::string	begin_distributed_transaction (const std::vector< std::shared_ptr< core::database_backend > > &participants)

async_result< bool >	commit_distributed_transaction (const std::string &transaction_id)

async_result< bool >	rollback_distributed_transaction (const std::string &transaction_id)

async_result< bool >	prepare_phase (const std::string &transaction_id)

async_result< bool >	commit_phase (const std::string &transaction_id)

saga_builder	create_saga ()

void	recover_transactions ()

std::vector< distributed_transaction >	get_active_transactions () const

Private Member Functions
async_result< bool >	two_phase_commit (const std::string &transaction_id)

void	cleanup_completed_transactions ()

Private Attributes
std::mutex	transactions_mutex_

std::unordered_map< std::string, distributed_transaction >	active_transactions_

std::shared_ptr< async_executor >	executor_

Detailed Description

Distributed transaction coordination.

Note: This class uses dependency injection pattern. Access via database_context::get_transaction_coordinator() (Sprint 3, Task 3.1).

Definition at line 652 of file async_operations.h.

Member Enumeration Documentation

◆ transaction_state

enum class database::async::transaction_coordinator::transaction_state

strong

Enumerator
active
preparing
prepared
committing
committed
aborting
aborted

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 655 of file async_operations.h.

                                     {
            active,
            preparing,
            prepared,
            committing,
            committed,
            aborting,
            aborted
        };

Constructor & Destructor Documentation

◆ transaction_coordinator()

database::async::transaction_coordinator::transaction_coordinator ( )

default

Default constructor - used by database_context.

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Member Function Documentation

◆ begin_distributed_transaction()

std::string database::async::transaction_coordinator::begin_distributed_transaction ( const std::vector< std::shared_ptr< core::database_backend > > & participants )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 892 of file async_operations.h.

    {
        static std::atomic<uint64_t> id_counter{0};
        auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        std::string txn_id = "txn-" + std::to_string(ms) + "-"
            + std::to_string(id_counter.fetch_add(1));
 
        std::lock_guard<std::mutex> lock(transactions_mutex_);
        distributed_transaction txn;
        txn.transaction_id = txn_id;
        txn.participants = participants;
        txn.state = transaction_state::active;
        txn.start_time = std::chrono::system_clock::now();
        txn.last_activity = txn.start_time;
        active_transactions_[txn_id] = std::move(txn);
        return txn_id;
    }

References active, active_transactions_, database::async::transaction_coordinator::distributed_transaction::last_activity, database::async::transaction_coordinator::distributed_transaction::participants, database::async::transaction_coordinator::distributed_transaction::start_time, database::async::transaction_coordinator::distributed_transaction::state, database::async::transaction_coordinator::distributed_transaction::transaction_id, and transactions_mutex_.

◆ cleanup_completed_transactions()

void database::async::transaction_coordinator::cleanup_completed_transactions ( )

inlineprivate

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1066 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(transactions_mutex_);
        for (auto it = active_transactions_.begin();
             it != active_transactions_.end();) {
            if (it->second.state == transaction_state::committed
                || it->second.state == transaction_state::aborted) {
                it = active_transactions_.erase(it);
            } else {
                ++it;
            }
        }
    }

References aborted, active_transactions_, committed, and transactions_mutex_.

Referenced by recover_transactions().

Here is the caller graph for this function:

◆ commit_distributed_transaction()

async_result< bool > database::async::transaction_coordinator::commit_distributed_transaction ( const std::string & transaction_id )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 994 of file async_operations.h.

    {
        return two_phase_commit(transaction_id);
    }

References two_phase_commit().

Here is the call graph for this function:

◆ commit_phase()

async_result< bool > database::async::transaction_coordinator::commit_phase ( const std::string & transaction_id )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 955 of file async_operations.h.

    {
        std::vector<std::shared_ptr<core::database_backend>> participants;
        {
            std::lock_guard<std::mutex> lock(transactions_mutex_);
            auto it = active_transactions_.find(transaction_id);
            if (it == active_transactions_.end()) {
                return make_error_result<bool>(
                    std::runtime_error("Transaction not found: " + transaction_id));
            }
            if (it->second.state != transaction_state::prepared) {
                return make_ready_result(false);
            }
            it->second.state = transaction_state::committing;
            it->second.last_activity = std::chrono::system_clock::now();
            participants = it->second.participants;
        }
 
        bool all_committed = true;
        for (const auto& participant : participants) {
            auto result = participant->commit_transaction();
            if (result.is_err()) {
                all_committed = false;
            }
        }
 
        {
            std::lock_guard<std::mutex> lock(transactions_mutex_);
            auto it = active_transactions_.find(transaction_id);
            if (it != active_transactions_.end()) {
                it->second.state = all_committed
                    ? transaction_state::committed
                    : transaction_state::aborted;
            }
        }
        return make_ready_result(all_committed);
    }

References aborted, active_transactions_, committed, committing, database::async::make_error_result(), database::async::make_ready_result(), prepared, and transactions_mutex_.

Referenced by two_phase_commit().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ create_saga()

saga_builder database::async::transaction_coordinator::create_saga ( )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1061 of file async_operations.h.

    {
        return saga_builder(*this);
    }

Referenced by TEST_F().

Here is the caller graph for this function:

◆ get_active_transactions()

std::vector< transaction_coordinator::distributed_transaction > database::async::transaction_coordinator::get_active_transactions ( ) const

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1050 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(transactions_mutex_);
        std::vector<distributed_transaction> result;
        result.reserve(active_transactions_.size());
        for (const auto& [id, txn] : active_transactions_) {
            result.push_back(txn);
        }
        return result;
    }

References active_transactions_, and transactions_mutex_.

◆ prepare_phase()

async_result< bool > database::async::transaction_coordinator::prepare_phase ( const std::string & transaction_id )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 912 of file async_operations.h.

    {
        std::vector<std::shared_ptr<core::database_backend>> participants;
        {
            std::lock_guard<std::mutex> lock(transactions_mutex_);
            auto it = active_transactions_.find(transaction_id);
            if (it == active_transactions_.end()) {
                return make_error_result<bool>(
                    std::runtime_error("Transaction not found: " + transaction_id));
            }
            it->second.state = transaction_state::preparing;
            it->second.last_activity = std::chrono::system_clock::now();
            participants = it->second.participants;
        }
 
        std::vector<std::shared_ptr<core::database_backend>> prepared;
        for (const auto& participant : participants) {
            auto result = participant->begin_transaction();
            if (result.is_err()) {
                for (const auto& p : prepared) {
                    p->rollback_transaction();
                }
                std::lock_guard<std::mutex> lock(transactions_mutex_);
                auto it = active_transactions_.find(transaction_id);
                if (it != active_transactions_.end()) {
                    it->second.state = transaction_state::aborted;
                }
                return make_ready_result(false);
            }
            prepared.push_back(participant);
        }
 
        {
            std::lock_guard<std::mutex> lock(transactions_mutex_);
            auto it = active_transactions_.find(transaction_id);
            if (it != active_transactions_.end()) {
                it->second.state = transaction_state::prepared;
            }
        }
        return make_ready_result(true);
    }

References aborted, active_transactions_, database::async::make_error_result(), database::async::make_ready_result(), prepared, preparing, and transactions_mutex_.

Referenced by two_phase_commit().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ recover_transactions()

void database::async::transaction_coordinator::recover_transactions ( )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1044 of file async_operations.h.

    {
        cleanup_completed_transactions();
    }

References cleanup_completed_transactions().

Here is the call graph for this function:

◆ rollback_distributed_transaction()

async_result< bool > database::async::transaction_coordinator::rollback_distributed_transaction ( const std::string & transaction_id )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1000 of file async_operations.h.

    {
        std::vector<std::shared_ptr<core::database_backend>> participants;
        {
            std::lock_guard<std::mutex> lock(transactions_mutex_);
            auto it = active_transactions_.find(transaction_id);
            if (it == active_transactions_.end()) {
                return make_error_result<bool>(
                    std::runtime_error("Transaction not found: " + transaction_id));
            }
            it->second.state = transaction_state::aborting;
            it->second.last_activity = std::chrono::system_clock::now();
            participants = it->second.participants;
        }
 
        bool all_rolled_back = true;
        for (const auto& participant : participants) {
            auto result = participant->rollback_transaction();
            if (result.is_err()) {
                all_rolled_back = false;
            }
        }
 
        {
            std::lock_guard<std::mutex> lock(transactions_mutex_);
            auto it = active_transactions_.find(transaction_id);
            if (it != active_transactions_.end()) {
                it->second.state = transaction_state::aborted;
            }
        }
        return make_ready_result(all_rolled_back);
    }

References aborted, aborting, active_transactions_, database::async::make_error_result(), database::async::make_ready_result(), and transactions_mutex_.

Here is the call graph for this function:

◆ two_phase_commit()

async_result< bool > database::async::transaction_coordinator::two_phase_commit ( const std::string & transaction_id )

inlineprivate

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1034 of file async_operations.h.

    {
        bool prepared = prepare_phase(transaction_id).get();
        if (!prepared) {
            return make_ready_result(false);
        }
        return commit_phase(transaction_id);
    }