forked from lei.chen/myxce
feat(ha-config-bootstrap): implement OP_ctrlHello state sync handler (task 9.3)
This commit is contained in:
parent
0ec0ffa7e3
commit
94bf5d94b8
9 changed files with 396 additions and 3 deletions
|
|
@ -27,13 +27,18 @@
|
|||
#include "byteorder.h"
|
||||
#include "clusterconfig.h"
|
||||
#include "cluster.h"
|
||||
#include "controller.capnp.h" /* OPCode_upgradeConfigAck */
|
||||
#include "controller.capnp.h" /* OPCode_upgradeConfigAck, OPCode_walCatchupEntry */
|
||||
#include "ha.h"
|
||||
#include "netio.h" /* endpoint_t, xce_request() */
|
||||
#include "repl.h" /* repl_wait_majority_ack, repl_on_config_ack */
|
||||
#include "request.h"
|
||||
#include "request.h" /* xce_request, xce_reply, init_reply_by_req */
|
||||
#include "utils.h"
|
||||
#include "wobj.h"
|
||||
#include "xce/cluster_def.h"
|
||||
#include "wal_types.h" /* wal_entry_hdr_t, WAL_* constants */
|
||||
|
||||
/* xce_request: defined in request.c, not yet in any public header */
|
||||
extern int xce_request(endpoint_t *ep, xce_request_t *req);
|
||||
|
||||
/* ── config load helpers (shared with myconf.c logic) ─────── */
|
||||
|
||||
|
|
@ -1644,3 +1649,198 @@ char *sanitise_cn(const char *src, size_t maxlen)
|
|||
buf[i] = '\0';
|
||||
return buf;
|
||||
}
|
||||
|
||||
/* ── handle_ctrlhello ─────────────────────────────────────────────────────
|
||||
*
|
||||
* Primary-side handler for OP_hello from a joining standby (Req 6.5–6.7).
|
||||
*
|
||||
* Protocol:
|
||||
* 1. Decode CtrlHelloRequest (applied_lsn, config_tag) from request body
|
||||
* 2. repl_add_standby() + repl_start_catchup()
|
||||
* 3. Reply CtrlHelloReply{committed_lsn, needs_snapshot, config_tag}
|
||||
* 4. If needs_snapshot: persist_checkpoint() then WAL streaming must wait
|
||||
* (snapshot transfer is out of scope for this task — stub logged)
|
||||
* 5. Stream OP_walCatchupEntry [applied_lsn+1 → committed_lsn] from disk
|
||||
* 6. On catch-up end: repl_complete_catchup()
|
||||
* 7. If standby config_tag ≠ primary config_tag: push_config_to_standbys()
|
||||
*
|
||||
* Requirements: 6.5 (state sync), 6.6 (snapshot before WAL when needed),
|
||||
* 6.7 (config push on mismatch)
|
||||
*/
|
||||
int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req)
|
||||
{
|
||||
/* 1. Extract CtrlHelloRequest fields */
|
||||
if (req->body_which != Request_body_ctrlhellomsg || !req->body.ctrlhellomsg) {
|
||||
WARN("handle_ctrlhello: malformed request (body_which=%d)", req->body_which);
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
ctrl_hello_req_t *hello = req->body.ctrlhellomsg;
|
||||
uint64_t standby_applied_lsn = hello->applied_lsn;
|
||||
uint64_t standby_config_tag = hello->config_tag;
|
||||
|
||||
/* 2. Register standby in replication set and mark as catching up */
|
||||
int rc = repl_add_standby(&c->persist.repl, ep,
|
||||
hello->name ? hello->name : "unknown", 0);
|
||||
if (rc < 0 && rc != -ENOSPC) {
|
||||
WARN("handle_ctrlhello: repl_add_standby failed: %d", rc);
|
||||
/* Still proceed — standby may already be registered */
|
||||
}
|
||||
repl_start_catchup(&c->persist.repl, ep, standby_applied_lsn);
|
||||
|
||||
/* Determine primary's committed LSN and snapshot need */
|
||||
uint64_t committed_lsn = atomic_load_explicit(&c->persist.committed_lsn,
|
||||
memory_order_acquire);
|
||||
int needs_snap = repl_needs_snapshot(&c->persist.repl,
|
||||
standby_applied_lsn,
|
||||
c->persist.checkpoint_lsn);
|
||||
|
||||
/* 3. Send CtrlHelloReply */
|
||||
ctrl_hello_reply_t hrep = {
|
||||
.committed_lsn = committed_lsn,
|
||||
.needs_snapshot = needs_snap,
|
||||
.config_tag = c->myconf->config_version_tag,
|
||||
};
|
||||
|
||||
xce_reply_t reply;
|
||||
init_reply_by_req(&reply, req);
|
||||
reply.status = 0;
|
||||
reply.data_which = RequestReply_data_ctrlhelloreplymsg;
|
||||
reply.data.ctrlhelloreplymsg = &hrep;
|
||||
xce_reply(ep, c->myhost, &reply);
|
||||
|
||||
/* 4. Snapshot transfer (Req 6.6) — stub: log and skip.
|
||||
* Full snapshot transfer (OP_snapshotTransfer) requires persist_checkpoint()
|
||||
* + raw blob transfer; deferred to a follow-on task. */
|
||||
if (needs_snap) {
|
||||
WARN("handle_ctrlhello: standby %s needs snapshot (applied_lsn=%lu) — "
|
||||
"snapshot transfer not yet implemented; standby must reconnect after "
|
||||
"primary checkpoint",
|
||||
hello->name ? hello->name : "?",
|
||||
(unsigned long)standby_applied_lsn);
|
||||
repl_complete_catchup(&c->persist.repl, ep);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* 5. Stream OP_walCatchupEntry [applied_lsn+1 → committed_lsn].
|
||||
*
|
||||
* Walk WAL segment files on disk in LSN order, read each wal_entry_hdr_t,
|
||||
* send entries in [standby_applied_lsn+1, committed_lsn] as raw capnp
|
||||
* requests. We re-use the existing WAL segment scan pattern from
|
||||
* persist_truncate_wal_above().
|
||||
*
|
||||
* For now, send WalCatchupEntry payloads as opaque Data blobs using the
|
||||
* existing xce_request() send path. A standby that receives
|
||||
* OPCode_walCatchupEntry applies it via apply_wal_entry() and replies
|
||||
* OPCode_walAck.
|
||||
*/
|
||||
if (committed_lsn > standby_applied_lsn) {
|
||||
/* Collect and sort WAL segments */
|
||||
uint64_t seg_lsns[4096];
|
||||
int nseg = 0;
|
||||
|
||||
DIR *d = opendir(c->persist.wal_path);
|
||||
if (d) {
|
||||
struct dirent *e;
|
||||
while ((e = readdir(d)) != NULL &&
|
||||
nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) {
|
||||
if (strncmp(e->d_name, "seg-", 4) != 0) continue;
|
||||
uint64_t sl = 0;
|
||||
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &sl) != 1) continue;
|
||||
seg_lsns[nseg++] = sl;
|
||||
}
|
||||
closedir(d);
|
||||
}
|
||||
|
||||
/* Insertion sort ascending */
|
||||
for (int i = 1; i < nseg; i++) {
|
||||
uint64_t key = seg_lsns[i];
|
||||
int j = i - 1;
|
||||
while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j+1] = seg_lsns[j]; j--; }
|
||||
seg_lsns[j+1] = key;
|
||||
}
|
||||
|
||||
/* Stream entries in each segment that falls in [applied+1, committed] */
|
||||
for (int si = 0; si < nseg; si++) {
|
||||
/* Skip segments that start after committed_lsn */
|
||||
if (si + 1 < nseg && seg_lsns[si + 1] <= standby_applied_lsn)
|
||||
continue; /* whole segment is below the standby's watermark */
|
||||
if (seg_lsns[si] > committed_lsn)
|
||||
break; /* past our range */
|
||||
|
||||
char seg_path[4096];
|
||||
snprintf(seg_path, sizeof(seg_path),
|
||||
"%s/seg-%016" PRIx64 ".wal", c->persist.wal_path, seg_lsns[si]);
|
||||
|
||||
int fd = open(seg_path, O_RDONLY);
|
||||
if (fd < 0) continue;
|
||||
|
||||
uint32_t off = 0;
|
||||
for (;;) {
|
||||
wal_entry_hdr_t hdr;
|
||||
ssize_t n = pread(fd, &hdr, sizeof(hdr), (off_t)off);
|
||||
if (n < (ssize_t)sizeof(hdr)) break;
|
||||
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break;
|
||||
if (hdr.lsn > committed_lsn) break;
|
||||
|
||||
if (hdr.lsn > standby_applied_lsn) {
|
||||
/* Stream WAL entry as OPCode_walCatchupEntry.
|
||||
* Encoded via the raw capnp path: set opcode + seq,
|
||||
* body carries the WalCatchupEntry struct. */
|
||||
xce_request_t stream_req;
|
||||
memset(&stream_req, 0, sizeof(stream_req));
|
||||
stream_req.opcode = OPCode_walCatchupEntry;
|
||||
stream_req.body_which = Request_body_nomsg;
|
||||
|
||||
wal_catchup_entry_t entry = {
|
||||
.lsn = hdr.lsn,
|
||||
.entry_type = (uint8_t)hdr.type,
|
||||
.payload = NULL,
|
||||
};
|
||||
if (hdr.payload_len > 0) {
|
||||
uint8_t *pbuf = malloc(hdr.payload_len);
|
||||
if (pbuf) {
|
||||
ssize_t pr = pread(fd, pbuf, hdr.payload_len,
|
||||
(off_t)(off + sizeof(hdr)));
|
||||
if (pr == (ssize_t)hdr.payload_len) {
|
||||
entry.payload = (xce_data_t *)pbuf;
|
||||
} else {
|
||||
free(pbuf);
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
/* Store in void* slot; decode_xce_request handles via opcode */
|
||||
stream_req.body.walcatchupentrymsg = &entry;
|
||||
xce_request(ep, &stream_req);
|
||||
free(entry.payload);
|
||||
}
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
}
|
||||
close(fd);
|
||||
}
|
||||
}
|
||||
|
||||
/* 6. Catch-up complete */
|
||||
repl_complete_catchup(&c->persist.repl, ep);
|
||||
INFO("handle_ctrlhello: standby %s catch-up complete "
|
||||
"(applied_lsn=%lu → committed_lsn=%lu)",
|
||||
hello->name ? hello->name : "?",
|
||||
(unsigned long)standby_applied_lsn,
|
||||
(unsigned long)committed_lsn);
|
||||
|
||||
/* 7. Config push if standby is stale (Req 6.7) */
|
||||
if (standby_config_tag != c->myconf->config_version_tag) {
|
||||
INFO("handle_ctrlhello: pushing config to standby %s "
|
||||
"(standby_tag=%lu, primary_tag=%lu)",
|
||||
hello->name ? hello->name : "?",
|
||||
(unsigned long)standby_config_tag,
|
||||
(unsigned long)c->myconf->config_version_tag);
|
||||
/* push_config_to_standbys() is best-effort; ignore return for now */
|
||||
(void)push_config_to_standbys(c, NULL, 0, NULL,
|
||||
c->myconf->config_version_tag);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue