coordinator.h File Reference

Ring consensus coordinator orchestration. More...

Go to the source code of this file.

Typedefs
typedef asciichat_error_t(*	consensus_election_func_t) (void *context, consensus_state_t *state)
	Election callback function type.
typedef asciichat_error_t(*	consensus_metrics_provider_fn) (void *, const uint8_t[16], participant_metrics_t *)

Functions
asciichat_error_t	consensus_coordinator_create (const uint8_t my_id[SESSION_ID_SIZE], const consensus_topology_t *topology, consensus_election_func_t election_func, void *election_context, consensus_coordinator_t **out_coordinator)
	Initialize ring consensus coordinator.
void	consensus_coordinator_destroy (consensus_coordinator_t *coordinator)
	Destroy coordinator and free resources.
asciichat_error_t	consensus_coordinator_process (consensus_coordinator_t *coordinator, uint32_t timeout_ms)
	Main orchestration loop - call periodically.
asciichat_error_t	consensus_coordinator_on_ring_members (consensus_coordinator_t *coordinator, const consensus_topology_t *new_topology)
	Update ring topology when participants change.
asciichat_error_t	consensus_coordinator_on_collection_start (consensus_coordinator_t *coordinator, uint32_t round_id, uint64_t deadline_ns)
	Handle STATS_COLLECTION_START packet.
asciichat_error_t	consensus_coordinator_on_stats_update (consensus_coordinator_t *coordinator, const uint8_t sender_id[SESSION_ID_SIZE], const participant_metrics_t *metrics, uint8_t num_metrics)
	Handle STATS_UPDATE packet (metrics relayed from another participant).
asciichat_error_t	consensus_coordinator_on_election_result (consensus_coordinator_t *coordinator, const uint8_t host_id[SESSION_ID_SIZE], const uint8_t backup_id[SESSION_ID_SIZE])
	Handle ELECTION_RESULT packet.
asciichat_error_t	consensus_coordinator_get_current_host (const consensus_coordinator_t *coordinator, uint8_t out_host_id[SESSION_ID_SIZE], uint8_t out_backup_id[SESSION_ID_SIZE])
	Get currently elected host.
consensus_state_machine_t	consensus_coordinator_get_state (const consensus_coordinator_t *coordinator)
	Get current coordinator state.
uint64_t	consensus_coordinator_time_until_next_round (const consensus_coordinator_t *coordinator)
	Get time until next round is due.
int	consensus_coordinator_get_metrics_count (const consensus_coordinator_t *coordinator)
	Get count of metrics collected in current round.
void	consensus_coordinator_set_metrics_provider (consensus_coordinator_t *, consensus_metrics_provider_fn, void *)

Detailed Description

Ring consensus coordinator orchestration.

Orchestrates the entire ring consensus flow:

• Periodic round scheduling (5 minute intervals)
• Metrics collection and relay around the ring
• Election computation by leader
• Result broadcasting and storage

The coordinator manages the state machine and calls appropriate functions based on the current consensus state and participant role.

Definition in file coordinator.h.

Typedef Documentation

consensus_election_func_t

typedef asciichat_error_t(* consensus_election_func_t) (void *context, consensus_state_t *state)

Election callback function type.

Called by coordinator when election needs to be computed by leader. The callback can run the deterministic election algorithm.

Parameters

context	Application context (passed to create)
state	State machine with collected metrics

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 36 of file coordinator.h.

consensus_metrics_provider_fn

typedef asciichat_error_t(* consensus_metrics_provider_fn) (void *, const uint8_t[16], participant_metrics_t *)

Session-owned observations; this callback must not perform blocking I/O.

Definition at line 191 of file coordinator.h.

Function Documentation

consensus_coordinator_create()

asciichat_error_t consensus_coordinator_create	(	const uint8_t	my_id[SESSION_ID_SIZE],
		const consensus_topology_t *	topology,
		consensus_election_func_t	election_func,
		void *	election_context,
		consensus_coordinator_t **	out_coordinator )

Initialize ring consensus coordinator.

Creates a new coordinator for managing consensus rounds. The coordinator will automatically schedule rounds every 5 minutes.

Parameters

my_id	My 16-byte UUID
topology	Topology handle (not owned, must remain valid during coordinator lifetime)
election_func	Callback to run election algorithm (leader only)
election_context	Application context passed to election_func
out_coordinator	Output coordinator handle (caller must destroy)

Returns

ASCIICHAT_OK on success, error code otherwise

Initialize ring consensus coordinator.

Definition at line 56 of file coordinator.c.

                                                                                          {
  if (!my_id || !topology || !election_func || !out_coordinator) {
    return SET_ERRNO(
        ERROR_INVALID_PARAM, "Invalid parameter: my_id=%p, topology=%p, election_func=%p, out_coordinator=%p",
        (const void *)my_id, (const void *)topology, (const void *)election_func, (const void *)out_coordinator);
  }
 
  consensus_coordinator_t *coordinator = SAFE_MALLOC(sizeof(consensus_coordinator_t), consensus_coordinator_t *);
  if (!coordinator) {
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate coordinator");
  }
 
  defer_named(release_coordinator, SAFE_FREE(coordinator));
  memset(coordinator, 0, sizeof(*coordinator));
  memcpy(coordinator->my_id, my_id, SESSION_ID_SIZE);
  coordinator->topology = topology;
  coordinator->election_func = election_func;
  coordinator->election_context = election_context;
 
  /* Initialize state machine */
  asciichat_error_t err = consensus_state_create(my_id, topology, &coordinator->state);
  if (err != ASCIICHAT_OK) {
    return err;
  }
 
  /* Initialize round scheduling */
  coordinator->last_round_start_ns = time_get_ns();
  coordinator->next_round_id = 1;
  coordinator->has_stored_result = false;
 
  /* Register coordinator with named registry */
  char coordinator_name[NETWORK_CONSENSUS_COORDINATOR_COORDINATOR_NAME_BUFFER_SIZE];
  snprintf(coordinator_name, sizeof(coordinator_name), "consensus_coordinator:node_%u", my_id[0]);
  NAMED_REGISTER(coordinator, coordinator_name, "consensus_coordinator_t", "0x%tx", NULL);
 
  /* Register child state with coordinator as parent */
  NAMED_REGISTER(coordinator->state, "state", "consensus_state_t", "0x%tx", (uintptr_t)(const void *)(coordinator));
 
  log_debug("Coordinator created for node %u, first round in 5 minutes", my_id[0]);
 
  *out_coordinator = coordinator;
  defer_cancel(release_coordinator);
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, consensus_state_create(), consensus_coordinator::election_context, consensus_coordinator::election_func, ERROR_INVALID_PARAM, ERROR_MEMORY, consensus_coordinator::has_stored_result, consensus_coordinator::last_round_start_ns, log_debug, consensus_coordinator::my_id, NAMED_REGISTER, NETWORK_CONSENSUS_COORDINATOR_COORDINATOR_NAME_BUFFER_SIZE, consensus_coordinator::next_round_id, SAFE_FREE, SAFE_MALLOC, SESSION_ID_SIZE, SET_ERRNO, consensus_coordinator::state, and consensus_coordinator::topology.

Referenced by session_consensus_create().

consensus_coordinator_destroy()

void consensus_coordinator_destroy

(

consensus_coordinator_t *

coordinator

)

Destroy coordinator and free resources.

Parameters

coordinator

Coordinator handle (safe to pass NULL)

Destroy coordinator and free resources.

Definition at line 107 of file coordinator.c.

                                                                         {
  if (!coordinator) {
    return;
  }
  consensus_state_destroy(coordinator->state);
  SAFE_FREE(coordinator);
}

References consensus_state_destroy(), SAFE_FREE, and consensus_coordinator::state.

Referenced by session_consensus_destroy().

consensus_coordinator_get_current_host()

asciichat_error_t consensus_coordinator_get_current_host	(	const consensus_coordinator_t *	coordinator,
		uint8_t	out_host_id[SESSION_ID_SIZE],
		uint8_t	out_backup_id[SESSION_ID_SIZE] )

Get currently elected host.

Retrieves the host and backup elected in the most recent successful round. If no round has completed yet, returns stored previous result if available, or an error if no result exists.

Parameters

coordinator	Coordinator handle
out_host_id	Output 16-byte host UUID (caller provides buffer)
out_backup_id	Output 16-byte backup UUID (caller provides buffer)

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 393 of file coordinator.c.

                                                                                                 {
  if (!coordinator || !out_host_id || !out_backup_id) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  /* Try to get from current state first */
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state == CONSENSUS_STATE_ELECTION_COMPLETE) {
    asciichat_error_t err = consensus_state_get_elected_host(coordinator->state, out_host_id);
    if (err == ASCIICHAT_OK) {
      err = consensus_state_get_elected_backup(coordinator->state, out_backup_id);
    }
    if (err == ASCIICHAT_OK) {
      return ASCIICHAT_OK;
    }
  }
 
  /* Fall back to stored result */
  if (coordinator->has_stored_result) {
    memcpy(out_host_id, coordinator->stored_host_id, SESSION_ID_SIZE);
    memcpy(out_backup_id, coordinator->stored_backup_id, SESSION_ID_SIZE);
    return ASCIICHAT_OK;
  }
 
  return SET_ERRNO(ERROR_INVALID_STATE, "No election result available");
}

References ASCIICHAT_OK, CONSENSUS_STATE_ELECTION_COMPLETE, consensus_state_get_current_state(), consensus_state_get_elected_backup(), consensus_state_get_elected_host(), ERROR_INVALID_PARAM, ERROR_INVALID_STATE, consensus_coordinator::has_stored_result, SESSION_ID_SIZE, SET_ERRNO, consensus_coordinator::state, consensus_coordinator::stored_backup_id, and consensus_coordinator::stored_host_id.

Referenced by session_consensus_get_elected_host(), and session_consensus_is_ready().

consensus_coordinator_get_metrics_count()

int consensus_coordinator_get_metrics_count

(

const consensus_coordinator_t *

coordinator

)

Get count of metrics collected in current round.

Used for testing and monitoring. Returns the number of metrics currently collected in the state machine.

Parameters

coordinator

Coordinator handle

Returns

Number of metrics collected, or -1 if invalid

Get count of metrics collected in current round.

Definition at line 453 of file coordinator.c.

                                                                                        {
  if (!coordinator || !coordinator->state) {
    return -1;
  }
  return consensus_state_get_metrics_count(coordinator->state);
}

References consensus_state_get_metrics_count(), and consensus_coordinator::state.

Referenced by session_consensus_get_metrics_count().

consensus_coordinator_get_state()

consensus_state_machine_t consensus_coordinator_get_state

(

const consensus_coordinator_t *

coordinator

)

Get current coordinator state.

Parameters

coordinator

Coordinator handle

Returns

Current state from internal state machine

Definition at line 425 of file coordinator.c.

                                                                                                      {
  if (!coordinator || !coordinator->state) {
    return CONSENSUS_STATE_FAILED;
  }
  return consensus_state_get_current_state(coordinator->state);
}

References CONSENSUS_STATE_FAILED, consensus_state_get_current_state(), and consensus_coordinator::state.

Referenced by session_consensus_get_state().

consensus_coordinator_on_collection_start()

asciichat_error_t consensus_coordinator_on_collection_start	(	consensus_coordinator_t *	coordinator,
		uint32_t	round_id,
		uint64_t	deadline_ns )

Handle STATS_COLLECTION_START packet.

Called when receiving a collection start message from the previous ring member. Prepares to collect metrics and transitions state machine.

Parameters

coordinator	Coordinator handle
round_id	Collection round ID
deadline_ns	Unix nanosecond deadline for this round

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 300 of file coordinator.c.

                                                                                  {
  if (!coordinator) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Coordinator is NULL");
  }
 
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state != CONSENSUS_STATE_IDLE) {
    return SET_ERRNO(ERROR_INVALID_STATE, "Cannot start collection, state is %d", current_state);
  }
 
  /* Start collection */
  asciichat_error_t err = consensus_state_start_collection(coordinator->state);
  if (err != ASCIICHAT_OK) {
    return err;
  }
 
  coordinator->next_round_id = round_id;
  coordinator->collection_deadline_ns = deadline_ns;
 
  /* Measure and add our metrics */
  err = measure_and_add_metrics(coordinator);
  if (err != ASCIICHAT_OK) {
    log_warn("Failed to measure metrics: %d", err);
  }
 
  uint64_t now_ns = time_get_ns();
  log_dev("Collection started: round_id=%u, deadline in %u seconds", round_id,
          (unsigned int)((deadline_ns > now_ns ? deadline_ns - now_ns : 0) / NS_PER_SEC_INT));
 
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, consensus_coordinator::collection_deadline_ns, consensus_state_get_current_state(), CONSENSUS_STATE_IDLE, consensus_state_start_collection(), ERROR_INVALID_PARAM, ERROR_INVALID_STATE, log_dev, log_warn, consensus_coordinator::next_round_id, NS_PER_SEC_INT, SET_ERRNO, and consensus_coordinator::state.

Referenced by session_consensus_on_collection_start().

consensus_coordinator_on_election_result()

asciichat_error_t consensus_coordinator_on_election_result	(	consensus_coordinator_t *	coordinator,
		const uint8_t	host_id[SESSION_ID_SIZE],
		const uint8_t	backup_id[SESSION_ID_SIZE] )

Handle ELECTION_RESULT packet.

Called when election result is received from leader. Verifies the election, stores the elected host/backup, and stores for fallback.

Parameters

coordinator	Coordinator handle
host_id	Elected host UUID
backup_id	Elected backup UUID

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 364 of file coordinator.c.

                                                                                                     {
  if (!coordinator || !host_id || !backup_id) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  /* Store the elected host and backup */
  memcpy(coordinator->stored_host_id, host_id, SESSION_ID_SIZE);
  memcpy(coordinator->stored_backup_id, backup_id, SESSION_ID_SIZE);
  coordinator->has_stored_result = true;
 
  log_info("Election result received: host=%u, backup=%u", host_id[0], backup_id[0]);
 
  /* Transition state back to IDLE */
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state == CONSENSUS_STATE_ELECTION_COMPLETE) {
    asciichat_error_t err = consensus_state_reset_to_idle(coordinator->state);
    if (err != ASCIICHAT_OK) {
      log_warn("Failed to reset state to IDLE: %d", err);
    }
  }
 
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, CONSENSUS_STATE_ELECTION_COMPLETE, consensus_state_get_current_state(), consensus_state_reset_to_idle(), ERROR_INVALID_PARAM, consensus_coordinator::has_stored_result, log_info, log_warn, SESSION_ID_SIZE, SET_ERRNO, consensus_coordinator::state, consensus_coordinator::stored_backup_id, and consensus_coordinator::stored_host_id.

Referenced by session_consensus_on_election_result().

consensus_coordinator_on_ring_members()

asciichat_error_t consensus_coordinator_on_ring_members	(	consensus_coordinator_t *	coordinator,
		const consensus_topology_t *	new_topology )

Update ring topology when participants change.

Called when the ring topology changes (participants join/leave). Updates internal topology reference and resets any in-progress round.

Parameters

coordinator	Coordinator handle
new_topology	New topology handle (not owned, must remain valid)

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 274 of file coordinator.c.

                                                                                                  {
  if (!coordinator || !new_topology) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  /* State also borrows the topology while idle. Build a complete replacement
   * before publishing either pointer so a rejected update leaves both valid. */
  consensus_state_t *replacement = NULL;
  asciichat_error_t err = consensus_state_create(coordinator->my_id, new_topology, &replacement);
  if (err != ASCIICHAT_OK)
    return err;
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state != CONSENSUS_STATE_IDLE)
    log_warn("Ring topology changed during round, resetting state");
  consensus_state_destroy(coordinator->state);
  coordinator->state = replacement;
  coordinator->topology = new_topology;
 
  log_info("Ring topology updated");
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, consensus_state_create(), consensus_state_destroy(), consensus_state_get_current_state(), CONSENSUS_STATE_IDLE, ERROR_INVALID_PARAM, log_info, log_warn, consensus_coordinator::my_id, SET_ERRNO, consensus_coordinator::state, and consensus_coordinator::topology.

Referenced by session_consensus_set_topology().

consensus_coordinator_on_stats_update()

asciichat_error_t consensus_coordinator_on_stats_update	(	consensus_coordinator_t *	coordinator,
		const uint8_t	sender_id[SESSION_ID_SIZE],
		const participant_metrics_t *	metrics,
		uint8_t	num_metrics )

Handle STATS_UPDATE packet (metrics relayed from another participant).

Called when receiving metrics from previous ring member. Adds metrics to state machine and forwards to next member.

Parameters

coordinator	Coordinator handle
sender_id	Who is sending this metrics update
metrics	Metrics to relay
num_metrics	Count of metrics in packet

Returns

ASCIICHAT_OK on success, error code otherwise

Handle STATS_UPDATE packet (metrics relayed from another participant).

Definition at line 336 of file coordinator.c.

                                                                                                                   {
  if (!coordinator || !sender_id || !metrics) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state != CONSENSUS_STATE_COLLECTING) {
    return SET_ERRNO(ERROR_INVALID_STATE, "Cannot accept metrics, state is %d", current_state);
  }
 
  /* Add all metrics to state */
  for (int i = 0; i < num_metrics; i++) {
    asciichat_error_t err = consensus_state_add_metrics(coordinator->state, &metrics[i]);
    if (err != ASCIICHAT_OK) {
      log_warn("Failed to add metric %d: %d", i, err);
    }
  }
 
  log_debug("Received %d metrics from sender %u", num_metrics, sender_id[0]);
 
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, consensus_state_add_metrics(), CONSENSUS_STATE_COLLECTING, consensus_state_get_current_state(), ERROR_INVALID_PARAM, ERROR_INVALID_STATE, log_debug, log_warn, SESSION_ID_SIZE, SET_ERRNO, and consensus_coordinator::state.

Referenced by session_consensus_on_stats_update().

consensus_coordinator_process()

asciichat_error_t consensus_coordinator_process	(	consensus_coordinator_t *	coordinator,
		uint32_t	timeout_ms )

Main orchestration loop - call periodically.

Checks if a new collection round should start and processes the current state. Should be called regularly (at least once per second) to ensure timely round scheduling.

Handles:

• Round scheduling (triggers every 5 minutes)
• Collection deadline enforcement (30 second deadline)
• Metrics measurement and relay
• Election computation (leader only)
• Result storage and broadcasting

Parameters

coordinator	Coordinator handle
timeout_ms	Timeout for operation in milliseconds

Returns

ASCIICHAT_OK on success, error code otherwise

Main orchestration loop - call periodically.

Definition at line 236 of file coordinator.c.

                                                                                                           {
  (void)timeout_ms; /* Currently unused, but kept for future use */
 
  if (!coordinator) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Coordinator is NULL");
  }
 
  uint64_t now_ns = time_get_ns();
 
  /* Check if we should start a new round */
  if (should_start_new_round(coordinator, now_ns)) {
    /* Only leader starts rounds */
    if (consensus_topology_am_leader(coordinator->topology)) {
      asciichat_error_t err = start_collection_round(coordinator, now_ns);
      if (err != ASCIICHAT_OK) {
        log_warn("Failed to start collection round: %d", err);
        return err;
      }
    }
  }
 
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
 
  /* Check for collection timeout */
  if (current_state == CONSENSUS_STATE_COLLECTING && has_collection_timed_out(coordinator, now_ns)) {
    log_warn("Collection round timed out, completing early");
    asciichat_error_t err = complete_collection(coordinator);
    if (err != ASCIICHAT_OK) {
      log_error("Failed to complete collection on timeout: %d", err);
    }
  }
 
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, CONSENSUS_STATE_COLLECTING, consensus_state_get_current_state(), consensus_topology_am_leader(), ERROR_INVALID_PARAM, log_error, log_warn, SET_ERRNO, consensus_coordinator::state, and consensus_coordinator::topology.

Referenced by session_consensus_process().

consensus_coordinator_set_metrics_provider()

void consensus_coordinator_set_metrics_provider	(	consensus_coordinator_t *	coordinator,
		consensus_metrics_provider_fn	provider,
		void *	context )

Definition at line 460 of file coordinator.c.

                                                                                                       {
  if (!coordinator)
    return;
  coordinator->metrics_provider = provider;
  coordinator->metrics_context = context;
}

References consensus_coordinator::metrics_context, and consensus_coordinator::metrics_provider.

Referenced by session_consensus_create().

consensus_coordinator_time_until_next_round()

uint64_t consensus_coordinator_time_until_next_round

(

const consensus_coordinator_t *

coordinator

)

Get time until next round is due.

Returns the number of nanoseconds until the next collection round is scheduled.

Parameters

coordinator

Coordinator handle

Returns

Nanoseconds until next round, or 0 if round is due now

Get time until next round is due.

Definition at line 435 of file coordinator.c.

                                                                                                 {
  if (!coordinator) {
    return 0;
  }
 
  uint64_t now_ns = time_get_ns();
  uint64_t next_round_ns = coordinator->last_round_start_ns + CONSENSUS_ROUND_INTERVAL_NS;
 
  if (now_ns >= next_round_ns) {
    return 0;
  }
 
  return next_round_ns - now_ns;
}

References CONSENSUS_ROUND_INTERVAL_NS, and consensus_coordinator::last_round_start_ns.

Referenced by session_consensus_time_until_next_round().