ab_gp_channel_negotiate.c

Go to the documentation of this file.

/* _POSIX_C_SOURCE 200809L: needed for select() below. */
#define _POSIX_C_SOURCE 200809L
 
#include "ab_gp_channel_negotiate.h"
#include "ab_gp_sibling_state.h"
#include "ab_gp_site_status.h"
#include "ab_gp_standby_policy.h"
#include "ab_gp_startup_timestamp.h"
#include "ab_gp_takeover_key.h"
 
#include <stdio.h>
#include <stdlib.h>
#include <strings.h>
#include <sys/select.h>
#include <sys/types.h>
#include <unistd.h>
 
#include "rte/oal/log/rte_log.h"
#include "rte/oal/memory/rte_mem_util.h"
#include "rte/oal/timer/rte_timer.h"
#include "rte/redundancy/site_role/rte_site_role.h"
#include "rte/redundancy/state_transfer/rte_state_transfer.h"
 
#include "../../common/common_config.h"
#include "ab_gp_channel.h"
#include "ab_gp_channel_payload.h"
#include "ab_gp_channel_util.h"
 
/* Lifetime of a takeover challenge (REQ-SITEROLE-015): long enough for the status line (every 5 s) to reach the
 * issuer and the token to come back, short enough that an unused token dies within minutes. */
#define AB_GP_TAKEOVER_CHALLENGE_MS 120000U
 
/* Startup timestamp exchange (ADR-040 step 5a): a STARTUP_TS request is re-sent this often while the sibling has
 * not answered (A's listen role has no peer until B connects). The read slice and the slice bound are in
 * ab_gp_startup_timestamp.h. */
#define AB_GP_STARTUP_TS_REQUEST_PERIOD_MS 100U
 
/* REQ-RBC-006/007: the negotiate-payload and AB_SAMPLE per-session encodings must agree byte-for-byte. */
typedef char negotiate_session_size_check_t
    [(RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE == RTE_EXAMPLE_SITE_EXTRA_PAYLOAD_SESSION_SIZE) ? 1 : -1];
/* The frame is the service's framing around GP's state region. */
typedef char negotiate_frame_layout_check_t
    [((RTE_EXAMPLE_SITE_EXTRA_PAYLOAD_SIZE ==
       (RTE_SITE_ROLE_FRAME_OVERHEAD + ((size_t)RTE_EXAMPLE_MAX_TRAINS * (size_t)RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE) +
        (size_t)RTE_EXAMPLE_DB_WIRE_SIZE)) &&
      (RTE_EXAMPLE_SITE_ROLE_VALID_MARKER == RTE_SITE_ROLE_VALID_MARKER) &&
      (RTE_EXAMPLE_SINGLE_MODE_FLAG_SINGLE == RTE_SITE_ROLE_SINGLE_MODE_SINGLE)) ? 1 : -1];

typedef struct
{
    ab_gp_train_session_t sessions[RTE_EXAMPLE_MAX_TRAINS];
    ab_gp_db_t            db;
} negotiate_state_t;
 
/* ---- state codec (rte_state_transfer registry; wire layout unchanged) ------------------------------------------- */
 
static rte_status_t encode_sessions_field(const void *data, uint8_t *out, size_t size)
{
    (void)size;
    ab_gp_channel_payload_encode_collector(ab_gp_train_payload_collector_id(), out,
                                            (const ab_gp_train_session_t *)data);
    return RTE_STATUS_OK;
}
 
static rte_status_t decode_sessions_field(void *data, const uint8_t *in, size_t size)
{
    (void)size;
    ab_gp_channel_payload_decode_collector(ab_gp_train_payload_collector_id(), in, (ab_gp_train_session_t *)data);
    return RTE_STATUS_OK;
}
 
static rte_status_t encode_db_field(const void *data, uint8_t *out, size_t size)
{
    uint32_t written = 0U;
    return ab_gp_db_encode((const ab_gp_db_t *)data, out, (uint32_t)size, &written);
}
 
static rte_status_t decode_db_field(void *data, const uint8_t *in, size_t size)
{
    return ab_gp_db_decode((ab_gp_db_t *)data, in, (uint32_t)size);
}

static void build_state_transfer_registry(rte_state_transfer_registry_t *registry, negotiate_state_t *state)
{
    (void)rte_state_transfer_registry_init(registry);
    (void)rte_state_transfer_register_field_ex(
        registry, "sessions", state->sessions,
        (size_t)RTE_EXAMPLE_MAX_TRAINS * (size_t)RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE, encode_sessions_field,
        decode_sessions_field);
    (void)rte_state_transfer_register_field_ex(registry, "db", &state->db, (size_t)RTE_EXAMPLE_DB_WIRE_SIZE,
                                                 encode_db_field, decode_db_field);
}

static rte_status_t encode_state(void *user, uint8_t *out, size_t size, bool live)
{
    const ab_gp_channel_context_t *ctx = (const ab_gp_channel_context_t *)user;
    negotiate_state_t state;
    rte_state_transfer_registry_t registry;
 
    rte_mem_set(&state, 0, sizeof(state));
    if (live)
    {
        rte_mem_copy(state.sessions, ctx->sessions, sizeof(state.sessions));
        rte_mem_copy(&state.db, &ctx->db, sizeof(state.db));
    }
    build_state_transfer_registry(&registry, &state);
    return rte_state_transfer_encode(&registry, out, size, NULL);
}

static bool receive_state(void *user, const uint8_t *in, size_t size, bool valid)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)user;
    negotiate_state_t decoded;
    rte_state_transfer_registry_t registry;
 
    rte_mem_set(&decoded, 0, sizeof(decoded));
    build_state_transfer_registry(&registry, &decoded);
    if (rte_state_transfer_decode(&registry, in, size) != RTE_STATUS_OK)
    {
        return false;
    }
    if (valid)
    {
        rte_mem_copy(ctx->neg_peer_extra.sessions, decoded.sessions, sizeof(ctx->neg_peer_extra.sessions));
        rte_mem_copy(&ctx->neg_peer_extra.db, &decoded.db, sizeof(ctx->neg_peer_extra.db));
    }
    return true;
}
 
/* ---- state transfer on promotion -------------------------------------------------------------------------------- */

static rte_example_transfer_policy_t read_transfer_policy(void)
{
    const char *env = getenv(RTE_EXAMPLE_TRANSFER_POLICY_ENV);
 
    if (env == NULL)
    {
        return RTE_EXAMPLE_TRANSFER_POLICY_CONTINUE;
    }
    if (strcasecmp(env, "restart") == 0)
    {
        return RTE_EXAMPLE_TRANSFER_POLICY_RESTART;
    }
    if (strcasecmp(env, "ask") == 0)
    {
        return RTE_EXAMPLE_TRANSFER_POLICY_ASK_USER;
    }
    return RTE_EXAMPLE_TRANSFER_POLICY_CONTINUE;
}

static bool ask_user_restart_or_continue(const ab_gp_channel_context_t *ctx, uint32_t transferred_cycle)
{
    fd_set read_fds;
    struct timeval timeout;
    int ready;
 
    RTE_EXAMPLE_LOG("[%s] *** ASK_USER: promoted to ONLINE with a transferred cycle=%u - restart (r) or continue (c)? "
           "[%ums to answer, default continue] ***\n",
           ctx->role_tag, transferred_cycle, (unsigned int)RTE_EXAMPLE_TRANSFER_ASK_USER_TIMEOUT_MS);
    rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TRANSFER",
                          "ASK_USER: awaiting operator restart/continue decision", NULL);
 
    FD_ZERO(&read_fds);
    FD_SET(STDIN_FILENO, &read_fds);
    timeout.tv_sec = (time_t)(RTE_EXAMPLE_TRANSFER_ASK_USER_TIMEOUT_MS / 1000U);
    timeout.tv_usec = (suseconds_t)((RTE_EXAMPLE_TRANSFER_ASK_USER_TIMEOUT_MS % 1000U) * 1000U);
 
    ready = select(STDIN_FILENO + 1, &read_fds, NULL, NULL, &timeout);
    if (ready > 0)
    {
        char line[8] = { 0 };
        ssize_t n = read(STDIN_FILENO, line, sizeof(line) - 1U);
 
        if (n > 0)
        {
            if ((line[0] == 'r') || (line[0] == 'R'))
            {
                RTE_EXAMPLE_LOG("[%s] ASK_USER: operator chose RESTART\n", ctx->role_tag);
                rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-",
                                      "TRANSFER", "ASK_USER: operator chose RESTART", NULL);
                return true;
            }
            RTE_EXAMPLE_LOG("[%s] ASK_USER: operator chose CONTINUE\n", ctx->role_tag);
            rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-",
                                  "TRANSFER", "ASK_USER: operator chose CONTINUE", NULL);
            return false;
        }
    }
    RTE_EXAMPLE_LOG("[%s] ASK_USER: no answer within %ums (EOF/no TTY/timeout) - defaulting to CONTINUE\n",
           ctx->role_tag, (unsigned int)RTE_EXAMPLE_TRANSFER_ASK_USER_TIMEOUT_MS);
    rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TRANSFER",
                          "ASK_USER: no answer in time - defaulting to CONTINUE", NULL);
    return false;
}

static void apply_state_transfer(ab_gp_channel_context_t *ctx, const rte_site_role_change_t *change)
{
    rte_example_transfer_policy_t policy;
    uint32_t applied_cycle;
 
    if (!change->snapshot_valid)
    {
        RTE_EXAMPLE_LOG("[%s] state transfer: no valid counterpart snapshot cached yet - starting cold (cycle 1, no sessions, no routes)\n",
               ctx->role_tag);
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TRANSFER",
                              "no valid counterpart snapshot cached - starting cold", NULL);
        ctx->cycle = 1U;
        rte_mem_set(ctx->sessions, 0, sizeof(ctx->sessions));
        rte_mem_set(&ctx->db, 0, sizeof(ctx->db));
        return;
    }
 
    policy = read_transfer_policy();
    applied_cycle = change->snapshot_cycle;
    switch (policy)
    {
        case RTE_EXAMPLE_TRANSFER_POLICY_RESTART:
            RTE_EXAMPLE_LOG("[%s] state transfer: policy=RESTART - renumbering from cycle 0 (transferred cycle was %u)\n",
                   ctx->role_tag, change->snapshot_cycle);
            applied_cycle = 0U;
            break;
        case RTE_EXAMPLE_TRANSFER_POLICY_ASK_USER:
            if (ask_user_restart_or_continue(ctx, change->snapshot_cycle))
            {
                applied_cycle = 0U;
            }
            break;
        case RTE_EXAMPLE_TRANSFER_POLICY_CONTINUE:
        default:
            RTE_EXAMPLE_LOG("[%s] state transfer: policy=CONTINUE - resuming from cycle %u\n", ctx->role_tag,
                   change->snapshot_cycle);
            break;
    }
 
    RTE_EXAMPLE_LOG("[%s] state transfer: applying {cycle=%u} + %u train session(s) + route db\n", ctx->role_tag,
           applied_cycle, (unsigned int)RTE_EXAMPLE_MAX_TRAINS);
    {
        char extra[48];
        (void)snprintf(extra, sizeof(extra), "cycle=%u policy=%d", applied_cycle, (int)policy);
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TRANSFER",
                              "applying transferred snapshot", extra);
    }
    /* ctx->cycle is "next cycle to run, 1-based"; a RESTART-policy transfer maps onto the same cold-start 1. */
    ctx->cycle = (applied_cycle > 0U) ? applied_cycle : 1U;
    /* Every "last seen at cycle N" stamp refers to the old numbering. The service resets its own link stamps
     * (peer_last_rx_cycle, standby_rx_cycle, standby_last_try_cycle in report_change()) but NOT the sibling
     * stamp (sibling_rx_cycle, REQ-SITEROLE-016): a CONTINUE renumbering upwards may count a frame taken before
     * the renumbering as reachable for one more cycle (the next execute takes a fresh frame and re-stamps it);
     * downwards the `cycle >= sibling_rx_cycle` guard of the service makes the old frame unreachable at once. */
    ctx->last_relay_rx_cycle = 0U;
    rte_mem_copy(ctx->sessions, ctx->neg_peer_extra.sessions, sizeof(ctx->sessions));
    rte_mem_copy(&ctx->db, &ctx->neg_peer_extra.db, sizeof(ctx->db));
}

static void on_role_change(void *user, const rte_site_role_change_t *change)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)user;
 
    ctx->is_online = change->online;
    if (change->kind == RTE_SITE_ROLE_CHANGE_PROMOTED)
    {
        /* A/B partner state is judged afresh from here: a STANDBY channel's flag can be stale, and one channel with the
         * flag set and its sibling without it is a site-state mismatch that reboots the pair right after a takeover. */
        ctx->single_mode = false;
        ctx->join_window_cycles_left = AB_GP_JOIN_ADOPT_CYCLES;
    }
    if (change->apply_state)
    {
        apply_state_transfer(ctx, change);
    }
}
 
static void trace(void *user, const char *line)
{
    (void)user;
    RTE_EXAMPLE_LOG("%s", line);
}
 
static void delay_ms(void *user, uint32_t ms)
{
    (void)user;
    ab_gp_channel_sleep_ms((unsigned int)ms);
}
 
/* ---- local facts ------------------------------------------------------------------------------------------------ */
 
static void local_facts(const ab_gp_channel_context_t *ctx, rte_site_role_local_t *out)
{
    out->faulted = ctx->own_faulted;
    out->single_mode = ctx->single_mode;
    out->own_ready = ab_gp_site_status_ready_for_online(ctx);
    /* Derived by rte_site_role from the sibling source (REQ-SITEROLE-016); unused with a source configured. */
    out->sibling_online = false;
    out->sibling_ready = false;
    out->sibling_reachable = false;
}

static bool sibling_source(void *user, rte_site_role_sibling_t *out)
{
    return ab_gp_sibling_state_take(&((ab_gp_channel_context_t *)user)->peer_site_state, out);
}
 
/* ADR-040 step 4b: with RTE_TAKEOVER_KEY_FILE set, only an authenticated token confirms a takeover (REQ-SITEROLE-015);
 * a set but unusable key file refuses to start, it never falls back to the unauthenticated path. */
static rte_status_t configure_takeover_key(ab_gp_channel_context_t *ctx, rte_site_role_config_t *cfg)
{
    const char *path = getenv(AB_GP_TAKEOVER_KEY_FILE_ENV);
    const char *site = site_config_name(ctx->site);
 
    if ((path == NULL) || (path[0] == '\0'))
    {
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site, 0U, ctx->role_tag, "-", "TAKEOVER",
                            "no RTE_TAKEOVER_KEY_FILE - dispatcher confirmation is unauthenticated (interim)", NULL);
        return RTE_STATUS_OK;
    }
    if (!ab_gp_takeover_key_load_file(path))
    {
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site, 0U, ctx->role_tag, "-", "TAKEOVER",
                            "RTE_TAKEOVER_KEY_FILE unreadable or malformed - refusing to start", NULL);
        return RTE_STATUS_INVALID_PARAM;
    }
    cfg->verify_mac = ab_gp_takeover_verify_mac;
    cfg->channel_id = (uint8_t)(ctx->is_role_a ? 0U : 1U);
    cfg->challenge_lifetime_cycles = (ctx->config.timers.cycle_period_ms != 0U)
                                         ? ((uint32_t)AB_GP_TAKEOVER_CHALLENGE_MS / ctx->config.timers.cycle_period_ms)
                                         : 0U; /* 0: the RTE default */
    rte_log_write_event(RTE_LOG_LEVEL_INFO, site, 0U, ctx->role_tag, "-", "TAKEOVER",
                        "dispatcher confirmation authenticated (RTE_TAKEOVER_KEY_FILE)", NULL);
    return RTE_STATUS_OK;
}
 
/* ---- site startup timestamp (ADR-040 step 5a) ------------------------------------------------------------------- */

static void send_startup_ts_frame(ab_gp_channel_context_t *ctx, bool is_reply, rte_timestamp_ms_t ts_ms)
{
    uint8_t frame[PEER_WIRE_FRAME_SIZE] = { 0 };
 
    frame[0] = PEER_MSG_KIND_STARTUP_TS;
    ab_gp_startup_timestamp_encode(&frame[1], is_reply, ts_ms);
    (void)ab_gp_channel_send_peer((ab_gp_channel_handle_t)ctx, frame, sizeof(frame));
}

static void exchange_startup_timestamp(ab_gp_channel_context_t *ctx)
{
    const char *site = site_config_name(ctx->site);
    const char *sibling_tag = ctx->is_role_a ? "B" : "A";
    rte_timestamp_ms_t own_now = 0U;
    rte_timestamp_ms_t now_ms = 0U;
    rte_timestamp_ms_t last_request_ms = 0U;
    rte_timestamp_ms_t sibling_ts = 0U;
    bool sibling_is_reply = false;
    bool have_sibling = false;
    bool request_pending = true;
    bool clock_failed = false;
    bool done = false;
    uint32_t slices_left = AB_GP_STARTUP_TS_MAX_SLICES;
    uint8_t in_bytes[PEER_WIRE_FRAME_SIZE];
    char sibling_text[24];
    char detail[48];
    char extra[96];
 
    if (rte_timer_now(&own_now) != RTE_STATUS_OK)
    {
        /* No clock: no exchange; startup_timestamp_ms stays 0 and rte_site_role_start() takes its own clock. */
        RTE_EXAMPLE_LOG("[%s] site startup timestamp: clock unavailable - no exchange with the sibling\n",
                        ctx->role_tag);
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site, 0U, ctx->role_tag, sibling_tag, "NEGOTIATION",
                            "startup timestamp: clock unavailable, no exchange", NULL);
        ctx->startup_timestamp_ms = 0U;
        return;
    }
 
    while (!done)
    {
        rte_duration_ms_t elapsed;
        rte_duration_ms_t remaining;
        rte_duration_ms_t slice_ms;
        rte_status_t status;
 
        if (slices_left == 0U)
        {
            done = true; /* hard bound: AB_GP_STARTUP_TS_MAX_SLICES slices read, the own clock is used */
            continue;
        }
        slices_left--;
        if (rte_timer_now(&now_ms) != RTE_STATUS_OK)
        {
            /* The clock failed after the exchange started: treated as the deadline, the own clock is used. */
            clock_failed = true;
            done = true;
            continue;
        }
        elapsed = (rte_duration_ms_t)(now_ms - own_now);
        if (elapsed >= (rte_duration_ms_t)AB_GP_STARTUP_TS_WAIT_MS)
        {
            done = true; /* timeout: the own clock is used */
            continue;
        }
        if (request_pending || ((now_ms - last_request_ms) >= (rte_timestamp_ms_t)AB_GP_STARTUP_TS_REQUEST_PERIOD_MS))
        {
            send_startup_ts_frame(ctx, false, own_now);
            last_request_ms = now_ms;
            request_pending = false;
        }
        remaining = (rte_duration_ms_t)AB_GP_STARTUP_TS_WAIT_MS - elapsed;
        slice_ms = (remaining > (rte_duration_ms_t)AB_GP_STARTUP_TS_READ_SLICE_MS)
                       ? (rte_duration_ms_t)AB_GP_STARTUP_TS_READ_SLICE_MS
                       : remaining;
 
        status = rte_channel_service_read(&ctx->com.peer_channel, in_bytes, sizeof(in_bytes), slice_ms);
        if (status == RTE_STATUS_OK)
        {
            if (in_bytes[0] == PEER_MSG_KIND_STARTUP_TS)
            {
                /* A malformed payload is dropped; the next request asks again. */
                have_sibling = ab_gp_startup_timestamp_decode(&in_bytes[1], sizeof(in_bytes) - 1U,
                                                              &sibling_is_reply, &sibling_ts);
                done = have_sibling;
            }
            else if (((in_bytes[0] == PEER_MSG_KIND_SAF_RB) || (in_bytes[0] == PEER_MSG_KIND_KM114_DIGEST)) &&
                     (ctx->deferred_peer_count < (uint8_t)AB_GP_DEFERRED_PEER_FRAMES))
            {
                /* One-off control frames, replayed by pre_execute (see poll_for_checkpoint_frame()). */
                rte_mem_copy(ctx->deferred_peer_frames[ctx->deferred_peer_count], in_bytes, sizeof(in_bytes));
                ctx->deferred_peer_count++;
            }
            else
            {
                /* other kinds are per-cycle data a later cycle re-sends */
            }
        }
        else if (status != RTE_STATUS_TIMEOUT)
        {
            /* No peer yet (A listens until B connects) or a read fault: keep the slice pacing, the deadline bounds
             * the loop either way. */
            ab_gp_channel_sleep_ms((unsigned int)slice_ms);
        }
        else
        {
            /* nothing arrived in this slice */
        }
    }
 
    if (have_sibling)
    {
        ctx->sibling_startup_timestamp_ms = sibling_ts;
        ctx->have_sibling_startup_timestamp = true;
        ctx->startup_timestamp_ms = ab_gp_startup_timestamp_agree(own_now, sibling_ts, true);
        if (!sibling_is_reply)
        {
            send_startup_ts_frame(ctx, true, ctx->startup_timestamp_ms);
        }
        (void)snprintf(sibling_text, sizeof(sibling_text), "%llu", (unsigned long long)sibling_ts);
        (void)snprintf(detail, sizeof(detail), "%s",
                       (ctx->startup_timestamp_ms == sibling_ts) ? "adopted sibling" : "own kept");
    }
    else
    {
        ctx->startup_timestamp_ms = own_now;
        (void)snprintf(sibling_text, sizeof(sibling_text), "%s", "none");
        (void)snprintf(detail, sizeof(detail), "sibling not reached within %u ms",
                       (unsigned int)AB_GP_STARTUP_TS_WAIT_MS);
    }
    RTE_EXAMPLE_LOG("[%s] site startup timestamp %llu ms (own %llu, sibling %s, %s)\n", ctx->role_tag,
                    (unsigned long long)ctx->startup_timestamp_ms, (unsigned long long)own_now, sibling_text, detail);
    (void)snprintf(extra, sizeof(extra), "agreed=%llu own=%llu sibling=%s",
                   (unsigned long long)ctx->startup_timestamp_ms, (unsigned long long)own_now, sibling_text);
    if (have_sibling)
    {
        rte_log_write_event(RTE_LOG_LEVEL_INFO, site, 0U, ctx->role_tag, sibling_tag, "NEGOTIATION",
                            "startup timestamp agreed", extra);
    }
    else if (clock_failed)
    {
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site, 0U, ctx->role_tag, sibling_tag, "NEGOTIATION",
                            "startup timestamp: clock failed during exchange, own used", extra);
    }
    else
    {
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site, 0U, ctx->role_tag, sibling_tag, "NEGOTIATION",
                            "startup timestamp: sibling not reached, own used", extra);
    }
}
 
void ab_gp_channel_negotiate_on_startup_ts_frame(ab_gp_channel_context_t *ctx, const uint8_t *payload, size_t size)
{
    bool is_reply = false;
    rte_timestamp_ms_t ts_ms = 0U;
 
    if ((ctx == NULL) || !ab_gp_startup_timestamp_decode(payload, size, &is_reply, &ts_ms))
    {
        return;
    }
    ctx->sibling_startup_timestamp_ms = ts_ms;
    ctx->have_sibling_startup_timestamp = true;
    if (!is_reply && (ctx->startup_timestamp_ms != 0U))
    {
        /* The sibling restarted and asks: it gets this channel's agreed value (own if the exchange timed out). A
         * channel whose clock was unavailable at start has no value (0, rte_site_role took its own clock) and does not
         * answer; the sibling then uses its own clock after its wait. */
        send_startup_ts_frame(ctx, true, ctx->startup_timestamp_ms);
    }
}
 
/* ---- GP entry points -------------------------------------------------------------------------------------------- */
 
rte_status_t ab_gp_channel_negotiate_init(ab_gp_channel_context_t *ctx)
{
    rte_site_role_config_t cfg;
    rte_status_t status;
    const bool west = (ctx->site == RTE_EXAMPLE_SITE_WEST);
 
    /* ADR-040 step 5a: both channels of the site present one timestamp to their inter-site tie-break. */
    exchange_startup_timestamp(ctx);
 
    (void)snprintf(ctx->site_role_peer_tag, sizeof(ctx->site_role_peer_tag), "%s/%s", ctx->is_role_a ? "A" : "B",
                   west ? "EAST" : "WEST");
    rte_mem_set(&cfg, 0, sizeof(cfg));
    cfg.own_id = (uint32_t)ctx->site;
    cfg.peer_id = west ? (uint32_t)RTE_EXAMPLE_SITE_EAST : (uint32_t)RTE_EXAMPLE_SITE_WEST;
    cfg.is_west = west;
    cfg.listen = west; /* WEST listens, EAST connects - same convention as every other link of this component */
    cfg.peer_host = west ? NULL
                         : site_config_host(ctx->is_role_a ? RTE_EXAMPLE_REMOTE_A_HOST_ENV : RTE_EXAMPLE_REMOTE_B_HOST_ENV);
    /* One fixed port per channel (not per site): A always RTE_EXAMPLE_A_NEGOTIATION_PORT, B always _B_. */
    cfg.port = ctx->is_role_a ? (uint16_t)RTE_EXAMPLE_A_NEGOTIATION_PORT : (uint16_t)RTE_EXAMPLE_B_NEGOTIATION_PORT;
    cfg.link_message_size = sizeof(rte_vital_message_t);
    cfg.ack_timeout_ms = (rte_duration_ms_t)ctx->config.timers.neg_ack_timeout_ms;
    cfg.peer_lost_timeout_ms = (rte_duration_ms_t)(
        (RTE_EXAMPLE_NEG_SEND_MISS_THRESHOLD * ctx->config.timers.neg_ack_timeout_ms) +
        ctx->config.timers.reconnect_attempt_timeout_ms + ctx->config.timers.cycle_period_ms);
    cfg.reconnect_attempt_ms = (rte_duration_ms_t)AB_GP_NEG_RECONNECT_ATTEMPT_MS;
    cfg.startup_timeout_ms = (rte_duration_ms_t)RTE_EXAMPLE_CONNECT_TIMEOUT_MS;
    cfg.link_poll_ms = (rte_duration_ms_t)RTE_EXAMPLE_LINK_TIMEOUT_MS;
    cfg.connect_retry_ms = 100U;
    cfg.connect_timeout_ms = 500U;
    cfg.send_miss_threshold = (uint8_t)RTE_EXAMPLE_NEG_SEND_MISS_THRESHOLD;
    cfg.standby_sync = (ctx->redundancy.standby_mode != RTE_STANDBY_MODE_COLD);
    cfg.startup_timestamp_ms = ctx->startup_timestamp_ms;
    cfg.frame_size = RTE_EXAMPLE_SITE_EXTRA_PAYLOAD_SIZE;
    cfg.tx_buf = ctx->site_role_tx;
    cfg.rx_buf = ctx->site_role_rx;
    cfg.role_tag = ctx->role_tag;
    cfg.peer_tag = ctx->site_role_peer_tag;
    cfg.site_name = site_config_name(ctx->site);
    cfg.sibling_tag = ctx->is_role_a ? "B" : "A";
    cfg.encode_state = encode_state;
    cfg.receive_state = receive_state;
    cfg.on_change = on_role_change;
    cfg.trace = trace;
    cfg.delay = delay_ms;
    cfg.user = ctx;
    /* ADR-040 step 5b: the service reads the A/B sibling's site state itself and derives online / ready / reachable;
     * 0 = RTE_SITE_ROLE_SIBLING_REACH_DEFAULT_CYCLES (3), the literal local_facts() used for reachability before. */
    cfg.sibling_source = sibling_source;
    cfg.sibling_reach_cycles = 0U;
 
    status = configure_takeover_key(ctx, &cfg);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
    status = rte_site_role_init(&ctx->site_role, &cfg);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
    status = rte_site_role_start(&ctx->site_role);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
    ctx->is_online = rte_site_role_is_online(&ctx->site_role);
    ctx->join_window_cycles_left = ctx->is_online ? AB_GP_JOIN_ADOPT_CYCLES : 0U;
    return RTE_STATUS_OK;
}
 
void ab_gp_channel_negotiate_execute(ab_gp_channel_context_t *ctx, uint32_t cycle)
{
    rte_site_role_local_t local;
 
    if (ctx->join_window_cycles_left > 0U)
    {
        ctx->join_window_cycles_left--;
    }
    local_facts(ctx, &local);
    rte_site_role_execute(&ctx->site_role, cycle, &local);
}
 
void ab_gp_channel_negotiate_service_link(ab_gp_channel_context_t *ctx)
{
    if (ctx != NULL)
    {
        rte_site_role_service_link(&ctx->site_role, ctx->cyc_cycle);
    }
}
 
void ab_gp_channel_negotiate_sync_standby(ab_gp_channel_context_t *ctx, uint32_t cycle)
{
    rte_site_role_local_t local;
 
    local_facts(ctx, &local);
    rte_site_role_sync_standby(&ctx->site_role, cycle, &local);
}
 
void ab_gp_channel_negotiate_on_takeover_confirm(ab_gp_channel_context_t *ctx, const char *payload)
{
    uint8_t token[RTE_SITE_ROLE_TOKEN_LEN];
 
    if (ctx == NULL)
    {
        return;
    }
    if (ab_gp_takeover_token_from_payload(payload, token))
    {
        (void)rte_site_role_on_takeover_token(&ctx->site_role, ctx->cyc_cycle, token, sizeof(token));
    }
    else
    {
        /* Not a token (the interim file trigger sends an empty payload): the unauthenticated path, which
         * rte_site_role refuses when a key is configured. */
        rte_site_role_on_takeover_confirm(&ctx->site_role, ctx->cyc_cycle);
    }
}
 
void ab_gp_channel_negotiate_flush_faulted(ab_gp_channel_context_t *ctx)
{
    rte_site_role_flush_faulted(&ctx->site_role, ctx->single_mode);
}
 
void ab_gp_channel_negotiate_shutdown(ab_gp_channel_context_t *ctx)
{
    rte_site_role_shutdown(&ctx->site_role);
}