diff --git a/src/controller/meson.build b/src/controller/meson.build index 08c3aa6..d9c7877 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -87,8 +87,10 @@ controller_unit_test_src = [ 'testsuite/test_ha_cluster.c', 'testsuite/test_ha_redirect.c', 'testsuite/test_p1_integration.c', + 'testsuite/test_repl_unit.c', 'persist.c', 'ha.c', + 'repl.c', ] controller_unit_test_deps = [ diff --git a/src/controller/repl.c b/src/controller/repl.c new file mode 100644 index 0000000..8d4dd22 --- /dev/null +++ b/src/controller/repl.c @@ -0,0 +1,249 @@ +/* + * repl.c — P2-1: Replicator lifecycle, slot management, and WAL trim guard. + * + * mylock_t is a plain pthread_mutex_t alias (see mylock.h). + * API: mylock_init / mylock_fini / mylock_lock / mylock_unlock + */ +#include +#include +#include + +#include "repl.h" +#include "mylock.h" + +/* ============================================================ + * Lifecycle + * ============================================================ */ + +int repl_init(replicator_t *r) +{ + if (r == NULL) { + return -EINVAL; + } + + memset(r, 0, sizeof(*r)); + + int rc = mylock_init(&r->lock); + if (rc != 0) { + return rc; + } + + rc = pthread_mutex_init(&r->ack_lock, NULL); + if (rc != 0) { + mylock_fini(&r->lock); + return -rc; + } + + rc = pthread_cond_init(&r->ack_cond, NULL); + if (rc != 0) { + pthread_mutex_destroy(&r->ack_lock); + mylock_fini(&r->lock); + return -rc; + } + + return 0; +} + +void repl_fini(replicator_t *r) +{ + if (r == NULL) { + return; + } + + pthread_cond_destroy(&r->ack_cond); + pthread_mutex_destroy(&r->ack_lock); + mylock_fini(&r->lock); +} + +/* ============================================================ + * Standby slot management + * ============================================================ */ + +int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port) +{ + if (r == NULL || ep == NULL || host == NULL) { + return -EINVAL; + } + + mylock_lock(&r->lock); + + if (r->nstandbys >= MAX_STANDBYS) { + mylock_unlock(&r->lock); + return -ENOSPC; + } + + /* Find first empty slot */ + int idx = -1; + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r->standbys[i].ep == NULL) { + idx = i; + break; + } + } + + if (idx < 0) { + /* Should not happen given nstandbys guard above, but be safe */ + mylock_unlock(&r->lock); + return -ENOSPC; + } + + standby_conn_t *s = &r->standbys[idx]; + memset(s, 0, sizeof(*s)); + s->ep = ep; + s->port = port; + strncpy(s->host, host, XCE_MAXNAME_LEN - 1); + s->host[XCE_MAXNAME_LEN - 1] = '\0'; + atomic_store(&s->acked_lsn, 0); + atomic_store(&s->in_catch_up, 0); + + r->nstandbys++; + + mylock_unlock(&r->lock); + return idx; +} + +void repl_remove_standby(replicator_t *r, repl_ep_t ep) +{ + if (r == NULL || ep == NULL) { + return; + } + + mylock_lock(&r->lock); + + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + if (s->ep == ep) { + s->last_disconnected_lsn = atomic_load(&s->acked_lsn); + s->last_disconnect_time = time(NULL); + s->ep = NULL; + r->nstandbys--; + break; + } + } + + mylock_unlock(&r->lock); +} + +void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s) +{ + if (r == NULL) { + return; + } + + mylock_lock(&r->lock); + + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + /* Only process disconnected (ep==NULL) slots that still hold a guard LSN */ + if (s->ep == NULL && s->last_disconnected_lsn != 0) { + /* snap_interval_s==0 means expire immediately regardless of wall clock */ + int expired = (snap_interval_s == 0) || + (now_s >= s->last_disconnect_time + (time_t)snap_interval_s); + if (expired) { + memset(s, 0, sizeof(*s)); + } + } + } + + mylock_unlock(&r->lock); +} + +/* ============================================================ + * ACK handling — call from io_worker read callback ONLY + * ============================================================ */ + +void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn) +{ + if (r == NULL || ep == NULL) { + return; + } + + mylock_lock(&r->lock); + + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + if (s->ep == ep) { + atomic_store(&s->acked_lsn, acked_lsn); + break; + } + } + + mylock_unlock(&r->lock); + + /* Signal any waiter (repl_wait_ack) */ + pthread_mutex_lock(&r->ack_lock); + pthread_cond_signal(&r->ack_cond); + pthread_mutex_unlock(&r->ack_lock); +} + +/* ============================================================ + * Replication send — stub in P2-1; wired in P2-4 + * ============================================================ */ + +int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, + const void *payload, uint32_t payload_len) +{ + (void)r; + (void)payload; + (void)payload_len; + + /* WAL_CHECKPOINT entries must never be replicated */ + assert(hdr != NULL); + assert(hdr->type != WAL_CHECKPOINT); + + /* Full implementation in P2-4 */ + return 0; +} + +/* ============================================================ + * Sync replication — stub in P2-1; wired in P2-3 + * ============================================================ */ + +int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms) +{ + (void)r; + (void)target_lsn; + (void)timeout_ms; + + /* Full implementation in P2-3 */ + return -ENODEV; +} + +/* ============================================================ + * WAL trim guard + * ============================================================ */ + +uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, + uint32_t snap_interval_s) +{ + (void)snap_interval_s; + + if (r == NULL) { + return checkpoint_lsn; + } + + uint64_t min_lsn = checkpoint_lsn; + + mylock_lock(&r->lock); + + for (int i = 0; i < MAX_STANDBYS; i++) { + const standby_conn_t *s = &r->standbys[i]; + + if (s->ep != NULL) { + /* Active standby: guard at its last acked LSN */ + uint64_t acked = atomic_load(&s->acked_lsn); + if (acked < min_lsn) { + min_lsn = acked; + } + } else if (s->last_disconnected_lsn != 0) { + /* Recently disconnected slot still holding a guard */ + if (s->last_disconnected_lsn < min_lsn) { + min_lsn = s->last_disconnected_lsn; + } + } + } + + mylock_unlock(&r->lock); + + return min_lsn; +} diff --git a/src/controller/repl.h b/src/controller/repl.h new file mode 100644 index 0000000..e135a0d --- /dev/null +++ b/src/controller/repl.h @@ -0,0 +1,59 @@ +#pragma once +/* + * repl.h — P2-1: Replicator structures and public API. + * Fan-out TCP replication via existing iomgr/SSL endpoint pipeline. + * ACKs processed in io_worker read callback (NOT vtaskmgr) to prevent deadlock. + */ +#include +#include +#include +#include +#include "config.h" +#include "mylock.h" +#include "persist.h" /* wal_entry_hdr_t */ + +/* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */ +typedef void* repl_ep_t; + +/* Per-standby connection slot */ +typedef struct { + repl_ep_t ep; /* NULL = slot empty */ + _Atomic uint64_t acked_lsn; /* Last LSN ACKed by this standby */ + _Atomic int in_catch_up; /* Non-zero while catching up */ + uint64_t join_applied_lsn; /* LSN standby reported on reconnect */ + char host[XCE_MAXNAME_LEN]; + uint32_t port; + uint64_t last_disconnected_lsn; + time_t last_disconnect_time; +} standby_conn_t; + +/* Replicator — embedded in persist_t or cluster_t (P2-4) */ +typedef struct { + standby_conn_t standbys[MAX_STANDBYS]; + int nstandbys; /* Active standby count */ + mylock_t lock; /* Protects standbys[], nstandbys */ + pthread_cond_t ack_cond; /* Signaled when any ACK arrives */ + pthread_mutex_t ack_lock; /* Guards ack_cond */ +} replicator_t; + +/* Lifecycle */ +int repl_init(replicator_t *r); +void repl_fini(replicator_t *r); + +/* Standby management */ +int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port); +void repl_remove_standby(replicator_t *r, repl_ep_t ep); +void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s); + +/* Replication send (stub in P2-1; wired in P2-4) */ +int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, + const void *payload, uint32_t payload_len); + +/* ACK handling — call from io_worker read callback ONLY */ +void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn); + +/* Sync replication (stub in P2-1; wired in P2-3) */ +int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms); + +/* WAL trim guard */ +uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, uint32_t snap_interval_s); diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index db8f201..8aba204 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -23,6 +23,7 @@ extern void test_ha_role(void); extern void test_ha_cluster(void); extern void test_ha_redirect(void); extern void test_p1_integration(void); +extern void test_repl_unit(void); void setUp(void) { return; } @@ -115,6 +116,7 @@ int main(int argc, char* argv[]) TEST(test_ha_cluster); TEST(test_ha_redirect); TEST(test_p1_integration); + TEST(test_repl_unit); return UNITY_END(); } diff --git a/src/controller/testsuite/test_repl_unit.c b/src/controller/testsuite/test_repl_unit.c new file mode 100644 index 0000000..a16258c --- /dev/null +++ b/src/controller/testsuite/test_repl_unit.c @@ -0,0 +1,85 @@ +#include +#include +#include +#include +/* Use the same Unity include pattern as other controller tests */ +#include +#include "repl.h" +#include "config.h" + +static void test_repl_init_zeroes_nstandbys(void) { + replicator_t r; + TEST_ASSERT_EQUAL_INT(0, repl_init(&r)); + TEST_ASSERT_EQUAL_INT(0, r.nstandbys); + repl_fini(&r); + printf("test_repl_init_zeroes_nstandbys: PASS\n"); +} + +static void test_repl_add_remove_standby(void) { + replicator_t r; + /* Use a dummy int as the endpoint pointer — repl.c only compares ep pointers */ + int dummy_ep = 0; + repl_init(&r); + int idx = repl_add_standby(&r, (void*)&dummy_ep, "standby1", 7001); + TEST_ASSERT_TRUE(idx >= 0); + TEST_ASSERT_EQUAL_INT(1, r.nstandbys); + repl_remove_standby(&r, (void*)&dummy_ep); + TEST_ASSERT_EQUAL_INT(0, r.nstandbys); + repl_fini(&r); + printf("test_repl_add_remove_standby: PASS\n"); +} + +static void test_repl_add_standby_max(void) { + replicator_t r; + int eps[MAX_STANDBYS + 1]; + repl_init(&r); + for (int i = 0; i < MAX_STANDBYS; i++) { + int rc = repl_add_standby(&r, (void*)&eps[i], "sb", 7000 + i); + TEST_ASSERT_TRUE(rc >= 0); + } + /* One over limit must fail */ + int rc = repl_add_standby(&r, (void*)&eps[MAX_STANDBYS], "sb_extra", 9999); + TEST_ASSERT_TRUE(rc < 0); + repl_fini(&r); + printf("test_repl_add_standby_max: PASS\n"); +} + +static void test_repl_trim_lsn_no_standbys(void) { + replicator_t r; + repl_init(&r); + uint64_t trim = repl_trim_lsn(&r, 42, 60); + TEST_ASSERT_EQUAL_UINT64(42, trim); + repl_fini(&r); + printf("test_repl_trim_lsn_no_standbys: PASS\n"); +} + +static void test_repl_trim_lsn_disconnected_slot_expiry(void) { + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); + repl_on_ack(&r, (void*)&dummy_ep, 10); + repl_remove_standby(&r, (void*)&dummy_ep); + + /* Immediately after disconnect: trim guard holds at last_disconnected_lsn=10 */ + uint64_t trim = repl_trim_lsn(&r, 50, 60); + TEST_ASSERT_EQUAL_UINT64(10, trim); + + /* After expiry (snap_interval_s=0 forces immediate expiry): trim == checkpoint */ + repl_expire_disconnect_slots(&r, 999999, 0); + trim = repl_trim_lsn(&r, 50, 0); + TEST_ASSERT_EQUAL_UINT64(50, trim); + + repl_fini(&r); + printf("test_repl_trim_lsn_disconnected_slot_expiry: PASS\n"); +} + +void test_repl_unit(void) { + UNITY_BEGIN(); + RUN_TEST(test_repl_init_zeroes_nstandbys); + RUN_TEST(test_repl_add_remove_standby); + RUN_TEST(test_repl_add_standby_max); + RUN_TEST(test_repl_trim_lsn_no_standbys); + RUN_TEST(test_repl_trim_lsn_disconnected_slot_expiry); + UNITY_END(); +}