forked from lei.chen/myxce
test(raft-ha-core-safety): machine-checkable walcompare report (task 1.1)
Per-member (lsn,term,pcrc,present) report + committed-vs-uncommitted divergence classification via walcompare_report()/--report. Soak signal (OK/exit-code) preserved. 4 new unit tests, all GREEN.
This commit is contained in:
parent
227f592250
commit
b028d25dbd
5 changed files with 453 additions and 120 deletions
|
|
@ -132,79 +132,202 @@ static int load_member(const walcompare_member_t* src, member_log_t* m, FILE* ou
|
|||
return 0;
|
||||
}
|
||||
|
||||
int walcompare(const walcompare_member_t* members, int n, FILE* out,
|
||||
walcompare_divergence_t* div)
|
||||
{
|
||||
if (!members || n < 2)
|
||||
return -EINVAL;
|
||||
if (n > WALCMP_MAX_MEMBERS)
|
||||
return -E2BIG;
|
||||
|
||||
/* Loaded cluster state shared by walcompare() and walcompare_report(). */
|
||||
typedef struct {
|
||||
member_log_t logs[WALCMP_MAX_MEMBERS];
|
||||
int loaded = 0;
|
||||
int rc = 0;
|
||||
int n;
|
||||
int loaded; /* members successfully loaded (for cleanup) */
|
||||
uint64_t boundary; /* highest committed LSN compared (committed edge) */
|
||||
uint64_t max_tip; /* highest recovered tip across members (tail edge) */
|
||||
} wc_cluster_t;
|
||||
|
||||
/* Stable status token for one member at one LSN. */
|
||||
static const char* wc_status(const member_log_t* m, uint64_t lsn)
|
||||
{
|
||||
if ((lsn < m->cap) && m->present[lsn])
|
||||
return "present";
|
||||
if (lsn <= m->committed_lsn)
|
||||
return "MISSING"; /* a committed entry this member SHOULD hold (R4.2) */
|
||||
return "beyond"; /* legitimately not held (above its committed prefix) */
|
||||
}
|
||||
|
||||
/* Load all members and compute the committed boundary + tail edge. On a load
|
||||
* error, frees what loaded, reports to `out`, returns a negative errno. */
|
||||
static int wc_load(const walcompare_member_t* members, int n, FILE* out,
|
||||
wc_cluster_t* cl)
|
||||
{
|
||||
memset(cl, 0, sizeof(*cl));
|
||||
cl->n = n;
|
||||
|
||||
for (int i = 0; i < n; i++) {
|
||||
rc = load_member(&members[i], &logs[i], out);
|
||||
int rc = load_member(&members[i], &cl->logs[i], out);
|
||||
if (rc != 0) {
|
||||
if (out)
|
||||
fprintf(out, "walcompare: cannot load member %d ('%s'): %s\n", i,
|
||||
members[i].path ? members[i].path : "(null)",
|
||||
strerror(-rc));
|
||||
goto done;
|
||||
for (int j = 0; j < cl->loaded; j++)
|
||||
member_log_free(&cl->logs[j]);
|
||||
return rc;
|
||||
}
|
||||
loaded = i + 1;
|
||||
cl->loaded = i + 1;
|
||||
}
|
||||
|
||||
/* Shared committed boundary = the minimum recovered tip across members:
|
||||
* the largest committed prefix every member can be expected to hold. A
|
||||
* member with an explicit committed_lsn keeps its own. */
|
||||
uint64_t shared = logs[0].recovered_tip;
|
||||
uint64_t shared = cl->logs[0].recovered_tip;
|
||||
for (int i = 1; i < n; i++)
|
||||
if (logs[i].recovered_tip < shared)
|
||||
shared = logs[i].recovered_tip;
|
||||
if (cl->logs[i].recovered_tip < shared)
|
||||
shared = cl->logs[i].recovered_tip;
|
||||
|
||||
uint64_t boundary = 0; /* highest committed LSN compared across the cluster */
|
||||
for (int i = 0; i < n; i++) {
|
||||
if (logs[i].committed_lsn == 0)
|
||||
logs[i].committed_lsn = shared;
|
||||
if (logs[i].committed_lsn > boundary)
|
||||
boundary = logs[i].committed_lsn;
|
||||
if (cl->logs[i].committed_lsn == 0)
|
||||
cl->logs[i].committed_lsn = shared;
|
||||
if (cl->logs[i].committed_lsn > cl->boundary)
|
||||
cl->boundary = cl->logs[i].committed_lsn;
|
||||
if (cl->logs[i].recovered_tip > cl->max_tip)
|
||||
cl->max_tip = cl->logs[i].recovered_tip;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void wc_unload(wc_cluster_t* cl)
|
||||
{
|
||||
for (int i = 0; i < cl->loaded; i++)
|
||||
member_log_free(&cl->logs[i]);
|
||||
}
|
||||
|
||||
/* One machine line for member `i` at `lsn`. Fixed field order, stable tokens. */
|
||||
static void wc_emit_member(const wc_cluster_t* cl, int i, uint64_t lsn,
|
||||
int committed, FILE* out)
|
||||
{
|
||||
const member_log_t* m = &cl->logs[i];
|
||||
int holds = (lsn < m->cap) && m->present[lsn];
|
||||
fprintf(out,
|
||||
"WALCMP member=%d lsn=%" PRIu64 " term=%" PRIu64
|
||||
" pcrc=%08x status=%s committed=%d\n",
|
||||
i, lsn, holds ? match_term(m->term[lsn]) : 0,
|
||||
holds ? m->pcrc[lsn] : 0u, wc_status(m, lsn), committed);
|
||||
}
|
||||
|
||||
/* Full per-member machine table over [1..max_tip]; committed=1 within the
|
||||
* committed boundary, 0 in the uncommitted tail. */
|
||||
static void wc_emit_table(const wc_cluster_t* cl, FILE* out)
|
||||
{
|
||||
for (uint64_t lsn = 1; lsn <= cl->max_tip; lsn++) {
|
||||
int committed = (lsn <= cl->boundary) ? 1 : 0;
|
||||
for (int i = 0; i < cl->n; i++)
|
||||
wc_emit_member(cl, i, lsn, committed, out);
|
||||
}
|
||||
}
|
||||
|
||||
/* Emit the classification + per-member identity for ONE diverging LSN. */
|
||||
static void wc_emit_divergence(const wc_cluster_t* cl, uint64_t lsn,
|
||||
int committed, int compared, int present,
|
||||
FILE* out)
|
||||
{
|
||||
fprintf(out,
|
||||
"WALCMP_DIVERGENCE lsn=%" PRIu64 " class=%s compared=%d present=%d\n",
|
||||
lsn, committed ? "committed" : "uncommitted", compared, present);
|
||||
for (int i = 0; i < cl->n; i++)
|
||||
wc_emit_member(cl, i, lsn, committed, out);
|
||||
}
|
||||
|
||||
/* Compare one committed LSN. Fills `div` (committed=1) and prints the human +
|
||||
* machine report when divergent. Returns 1 if divergent, else 0. */
|
||||
static int wc_compare_committed(const wc_cluster_t* cl,
|
||||
const walcompare_member_t* members, uint64_t lsn,
|
||||
FILE* out, walcompare_divergence_t* div)
|
||||
{
|
||||
int have = 0, present = 0, have_ref = 0, divergent = 0;
|
||||
uint64_t ref_term = 0;
|
||||
uint32_t ref_pcrc = 0;
|
||||
|
||||
for (int i = 0; i < cl->n; i++) {
|
||||
if (lsn > cl->logs[i].committed_lsn)
|
||||
continue; /* beyond member i's committed prefix */
|
||||
have++;
|
||||
const member_log_t* m = &cl->logs[i];
|
||||
int holds = (lsn < m->cap) && m->present[lsn];
|
||||
if (!holds) {
|
||||
divergent = 1; /* member should hold this committed LSN but doesn't */
|
||||
continue;
|
||||
}
|
||||
present++;
|
||||
uint64_t mt = match_term(m->term[lsn]);
|
||||
uint32_t mp = m->pcrc[lsn];
|
||||
if (!have_ref) {
|
||||
ref_term = mt;
|
||||
ref_pcrc = mp;
|
||||
have_ref = 1;
|
||||
} else if (mt != ref_term || mp != ref_pcrc) {
|
||||
divergent = 1;
|
||||
}
|
||||
}
|
||||
|
||||
if (out)
|
||||
if (have < 2 || !divergent)
|
||||
return 0;
|
||||
|
||||
if (div) {
|
||||
memset(div, 0, sizeof(*div));
|
||||
div->lsn = lsn;
|
||||
div->n_members = cl->n;
|
||||
div->committed = 1;
|
||||
for (int i = 0; i < cl->n && i < 16; i++) {
|
||||
const member_log_t* m = &cl->logs[i];
|
||||
int holds = (lsn <= m->committed_lsn) && (lsn < m->cap) &&
|
||||
m->present[lsn];
|
||||
div->member_present[i] = holds ? 1 : 0;
|
||||
div->member_term[i] = holds ? match_term(m->term[lsn]) : 0;
|
||||
div->member_pcrc[i] = holds ? m->pcrc[lsn] : 0;
|
||||
}
|
||||
}
|
||||
if (out) {
|
||||
fprintf(out,
|
||||
"== walcompare: %d members, committed boundary up to LSN %" PRIu64
|
||||
" ==\n",
|
||||
n, boundary);
|
||||
|
||||
/* Walk every committed LSN. At each LSN, compare the (term, pcrc) of the
|
||||
* members whose own committed boundary covers it and that hold the entry.
|
||||
* A member that does not hold a committed LSN it SHOULD (boundary covers it
|
||||
* but the entry is absent) is itself a committed-prefix divergence: a
|
||||
* committed entry is missing (R4.2 / R6.4). */
|
||||
for (uint64_t lsn = 1; lsn <= boundary; lsn++) {
|
||||
int have = 0; /* members whose boundary covers this LSN */
|
||||
int present = 0; /* of those, how many actually hold it */
|
||||
uint64_t ref_term = 0;
|
||||
uint32_t ref_pcrc = 0;
|
||||
int have_ref = 0;
|
||||
int divergent = 0;
|
||||
|
||||
for (int i = 0; i < n; i++) {
|
||||
if (lsn > logs[i].committed_lsn)
|
||||
continue; /* this LSN is beyond member i's committed prefix */
|
||||
have++;
|
||||
|
||||
int holds = (lsn < logs[i].cap) && logs[i].present[lsn];
|
||||
if (!holds) {
|
||||
/* member i should hold this committed LSN but does not */
|
||||
divergent = 1;
|
||||
"*** DIVERGENCE at committed LSN %" PRIu64
|
||||
" (compared %d, present %d) ***\n",
|
||||
lsn, have, present);
|
||||
for (int i = 0; i < cl->n; i++) {
|
||||
if (lsn > cl->logs[i].committed_lsn) {
|
||||
fprintf(out,
|
||||
" member %d ('%s'): LSN beyond its committed prefix"
|
||||
" (committed=%" PRIu64 ")\n",
|
||||
i, members[i].path, cl->logs[i].committed_lsn);
|
||||
continue;
|
||||
}
|
||||
int holds = (lsn < cl->logs[i].cap) && cl->logs[i].present[lsn];
|
||||
if (holds)
|
||||
fprintf(out, " member %d ('%s'): term=%" PRIu64 " pcrc=%08x\n",
|
||||
i, members[i].path, match_term(cl->logs[i].term[lsn]),
|
||||
cl->logs[i].pcrc[lsn]);
|
||||
else
|
||||
fprintf(out, " member %d ('%s'): MISSING committed entry\n", i,
|
||||
members[i].path);
|
||||
}
|
||||
wc_emit_divergence(cl, lsn, 1, have, present, out);
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* Scan the uncommitted tail (boundary < lsn <= max_tip): emit a NON-fatal
|
||||
* class=uncommitted classification for any LSN where >=2 members hold the entry
|
||||
* but disagree on (term, pcrc). Never changes the verdict. */
|
||||
static void wc_scan_uncommitted(const wc_cluster_t* cl, FILE* out)
|
||||
{
|
||||
for (uint64_t lsn = cl->boundary + 1; lsn <= cl->max_tip; lsn++) {
|
||||
int present = 0, have_ref = 0, divergent = 0;
|
||||
uint64_t ref_term = 0;
|
||||
uint32_t ref_pcrc = 0;
|
||||
|
||||
for (int i = 0; i < cl->n; i++) {
|
||||
const member_log_t* m = &cl->logs[i];
|
||||
int holds = (lsn < m->cap) && m->present[lsn];
|
||||
if (!holds)
|
||||
continue;
|
||||
present++;
|
||||
uint64_t mt = match_term(logs[i].term[lsn]);
|
||||
uint32_t mp = logs[i].pcrc[lsn];
|
||||
uint64_t mt = match_term(m->term[lsn]);
|
||||
uint32_t mp = m->pcrc[lsn];
|
||||
if (!have_ref) {
|
||||
ref_term = mt;
|
||||
ref_pcrc = mp;
|
||||
|
|
@ -213,60 +336,69 @@ int walcompare(const walcompare_member_t* members, int n, FILE* out,
|
|||
divergent = 1;
|
||||
}
|
||||
}
|
||||
|
||||
if (have < 2)
|
||||
continue; /* nothing to compare at this LSN */
|
||||
if (!divergent)
|
||||
if (present < 2 || !divergent)
|
||||
continue;
|
||||
|
||||
/* First divergent committed LSN — report and fail. */
|
||||
if (div) {
|
||||
memset(div, 0, sizeof(*div));
|
||||
div->lsn = lsn;
|
||||
div->n_members = n;
|
||||
for (int i = 0; i < n && i < 16; i++) {
|
||||
int holds = (lsn <= logs[i].committed_lsn) &&
|
||||
(lsn < logs[i].cap) && logs[i].present[lsn];
|
||||
div->member_present[i] = holds ? 1 : 0;
|
||||
div->member_term[i] = holds ? match_term(logs[i].term[lsn]) : 0;
|
||||
div->member_pcrc[i] = holds ? logs[i].pcrc[lsn] : 0;
|
||||
}
|
||||
}
|
||||
if (out) {
|
||||
fprintf(out,
|
||||
"*** DIVERGENCE at committed LSN %" PRIu64
|
||||
" (compared %d, present %d) ***\n",
|
||||
lsn, have, present);
|
||||
for (int i = 0; i < n; i++) {
|
||||
if (lsn > logs[i].committed_lsn) {
|
||||
fprintf(out,
|
||||
" member %d ('%s'): LSN beyond its committed prefix"
|
||||
" (committed=%" PRIu64 ")\n",
|
||||
i, members[i].path, logs[i].committed_lsn);
|
||||
continue;
|
||||
}
|
||||
int holds = (lsn < logs[i].cap) && logs[i].present[lsn];
|
||||
if (holds)
|
||||
fprintf(out,
|
||||
" member %d ('%s'): term=%" PRIu64 " pcrc=%08x\n", i,
|
||||
members[i].path, match_term(logs[i].term[lsn]),
|
||||
logs[i].pcrc[lsn]);
|
||||
else
|
||||
fprintf(out,
|
||||
" member %d ('%s'): MISSING committed entry\n", i,
|
||||
members[i].path);
|
||||
}
|
||||
}
|
||||
rc = WALCMP_DIVERGENT;
|
||||
goto done;
|
||||
wc_emit_divergence(cl, lsn, 0, present, present, out);
|
||||
}
|
||||
}
|
||||
|
||||
/* Shared driver for walcompare() (full_report=0) and walcompare_report()
|
||||
* (full_report=1). */
|
||||
static int wc_run(const walcompare_member_t* members, int n, FILE* out,
|
||||
walcompare_divergence_t* div, int full_report)
|
||||
{
|
||||
if (!members || n < 2)
|
||||
return -EINVAL;
|
||||
if (n > WALCMP_MAX_MEMBERS)
|
||||
return -E2BIG;
|
||||
|
||||
wc_cluster_t cl;
|
||||
int rc = wc_load(members, n, out, &cl);
|
||||
if (rc != 0)
|
||||
return rc;
|
||||
|
||||
if (out)
|
||||
fprintf(out,
|
||||
"== walcompare: %d members, committed boundary up to LSN %" PRIu64
|
||||
" ==\n",
|
||||
n, cl.boundary);
|
||||
|
||||
if (full_report && out)
|
||||
wc_emit_table(&cl, out);
|
||||
|
||||
/* Walk every committed LSN. At each LSN, compare the (term, pcrc) of the
|
||||
* members whose own committed boundary covers it and that hold the entry.
|
||||
* A member that does not hold a committed LSN it SHOULD is a committed-
|
||||
* prefix divergence: a committed entry is missing (R4.2 / R6.4). */
|
||||
rc = WALCMP_OK;
|
||||
for (uint64_t lsn = 1; lsn <= cl.boundary; lsn++) {
|
||||
if (wc_compare_committed(&cl, members, lsn, out, div)) {
|
||||
rc = WALCMP_DIVERGENT; /* first committed divergence is the verdict */
|
||||
goto done;
|
||||
}
|
||||
}
|
||||
|
||||
/* The committed prefix is identical. In full-report mode also classify any
|
||||
* uncommitted-tail divergence (informational, never fatal). */
|
||||
if (full_report && out)
|
||||
wc_scan_uncommitted(&cl, out);
|
||||
|
||||
if (out)
|
||||
fprintf(out, "walcompare: OK — committed prefixes identical\n");
|
||||
rc = WALCMP_OK;
|
||||
|
||||
done:
|
||||
for (int i = 0; i < loaded; i++)
|
||||
member_log_free(&logs[i]);
|
||||
wc_unload(&cl);
|
||||
return rc;
|
||||
}
|
||||
|
||||
int walcompare(const walcompare_member_t* members, int n, FILE* out,
|
||||
walcompare_divergence_t* div)
|
||||
{
|
||||
return wc_run(members, n, out, div, 0);
|
||||
}
|
||||
|
||||
int walcompare_report(const walcompare_member_t* members, int n, FILE* out,
|
||||
walcompare_divergence_t* div)
|
||||
{
|
||||
return wc_run(members, n, out, div, 1);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue