Struct PgElection

pub struct PgElection {
    leader_value: String,
    pg_client: RwLock<ElectionPgClient>,
    is_leader: AtomicBool,
    leader_infancy: AtomicBool,
    leader_watcher: Sender<LeaderChangeMessage>,
    store_key_prefix: String,
    candidate_lease_ttl: Duration,
    meta_lease_ttl: Duration,
    sql_set: ElectionSqlSet,
}

PostgreSql implementation of Election.

Fields

leader_value: String

pg_client: RwLock<ElectionPgClient>

is_leader: AtomicBool

leader_infancy: AtomicBool

leader_watcher: Sender<LeaderChangeMessage>

store_key_prefix: String

candidate_lease_ttl: Duration

meta_lease_ttl: Duration

sql_set: ElectionSqlSet

Implementations

impl PgElection

async fn maybe_init_client(&self) -> Result<()>

pub async fn with_pg_client( leader_value: String, pg_client: ElectionPgClient, store_key_prefix: String, candidate_lease_ttl: Duration, meta_lease_ttl: Duration, table_name: &str, lock_id: u64, ) -> Result<ElectionRef>

fn election_key(&self) -> String

fn candidate_root(&self) -> String

fn candidate_key(&self) -> String

impl PgElection

async fn get_value_with_lease(&self, key: &str) -> Result<Option<Lease>>

Returns value, expire time and current time. If with_origin is true, the origin string is also returned.

async fn get_value_with_lease_by_prefix( &self, key_prefix: &str, ) -> Result<(Vec<(String, Timestamp)>, Timestamp)>

Returns all values and expire time with the given key prefix. Also returns the current time.

async fn update_value_with_lease( &self, key: &str, prev: &str, updated: &str, lease_ttl: Duration, ) -> Result<()>

async fn put_value_with_lease( &self, key: &str, value: &str, lease_ttl: Duration, ) -> Result<bool>

Returns true if the insertion is successful

async fn delete_value(&self, key: &str) -> Result<bool>

Returns true if the deletion is successful. Caution: Should only delete the key if the lease is expired.

async fn leader_action(&self) -> Result<()>

Handles the actions of a leader in the election process.

This function performs the following checks and actions:

• Case 1: If the current instance believes it is the leader from the previous term, it attempts to renew the lease. It checks if the lease is still valid and either renews it or steps down if it has expired.

  • Case 1.1: If the instance is still the leader and the lease is valid, it renews the lease by updating the value associated with the election key.
  • Case 1.2: If the instance is still the leader but the lease has expired, it logs a warning and steps down, initiating a new campaign for leadership.
  • Case 1.3: If the instance is not the leader (which is a rare scenario), it logs a warning indicating that it still holds the lock and steps down to re-initiate the campaign. This may happen if the leader has failed to renew the lease and the session has expired, and recovery after a period of time during which other leaders have been elected and stepped down.
  • Case 1.4: If no lease information is found, it also steps down and re-initiates the campaign.

• Case 2: If the current instance is not leader previously, it calls the elected method as a newly elected leader.

async fn follower_action(&self) -> Result<()>

Handles the actions of a follower in the election process.

This function performs the following checks and actions:

• Case 1: If the current instance believes it is the leader from the previous term, it steps down without deleting the key.
• Case 2: If the current instance is not the leader but the lease has expired, it raises an error to re-initiate the campaign. If the leader failed to renew the lease, its session will expire and the lock will be released.
• Case 3: If all checks pass, the function returns without performing any actions.

async fn step_down(&self) -> Result<()>

Step down the leader. The leader should delete the key and notify the leader watcher.

DO NOT check if the deletion is successful, since the key may be deleted by others elected.

Caution:

Should only step down while holding the advisory lock.

async fn step_down_without_lock(&self) -> Result<()>

Still consider itself as the leader locally but failed to acquire the lock. Step down without deleting the key.

async fn elected(&self) -> Result<()>

Elected as leader. The leader should put the key and notify the leader watcher. Caution: Should only elected while holding the advisory lock.

Trait Implementations

impl Election for PgElection

fn campaign<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Attempts to acquire leadership by executing a campaign. This function continuously checks if the current instance can become the leader by acquiring an advisory lock in the PostgreSQL database.

The function operates in a loop, where it:

1. Waits for a predefined interval before attempting to acquire the lock again.
2. Executes the CAMPAIGN SQL query to try to acquire the advisory lock.
3. Checks the result of the query:
   • If the lock is successfully acquired (result is true), it calls the leader_action method to perform actions as the leader.
   • If the lock is not acquired (result is false), it calls the follower_action method to perform actions as a follower.

type Leader = LeaderValue

fn is_leader(&self) -> bool

Returns true if current node is the leader.

fn in_leader_infancy(&self) -> bool

When a new leader is born, it may need some initialization operations (asynchronous), this method tells us when these initialization operations can be performed.

fn register_candidate<'life0, 'life1, 'async_trait>( &'life0 self, node_info: &'life1 MetasrvNodeInfo, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Registers a candidate for the election.

fn all_candidates<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<MetasrvNodeInfo>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Gets all candidates in the election.

fn reset_campaign<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Resets the campaign.

fn leader<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Self::Leader>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns the leader value for the current election.

fn resign<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Releases election leadership so other campaigners may acquire leadership on the election.

fn subscribe_leader_change(&self) -> Receiver<LeaderChangeMessage>