feat(libxce-failover): reconnect recovery — ctx ERR->RUN reset + stale wbuf drain (fixes post-failover EINVAL) (task 3.2)

src/include/mylock.h

@@ -35,4 +35,5 @@ extern void cbuf_fini(cbuf_t*);
 extern int  cbuf_put(cbuf_t*, void*);
 extern int  cbuf_tryput(cbuf_t*, void*);
 extern int  cbuf_get(cbuf_t*, void*);
+extern int  cbuf_tryget(cbuf_t*, void*);
 #endif

src/lib/client_failover.c

@@ -159,6 +159,36 @@ int xce_failover_rotate(api_ctx_t *ac)
     return rc < 0 ? rc : 0;
 }
 
+/* ════════════════════════════════════════════════════════════════════════
+ * Reconnect recovery — context-state reset (CI2) + stale-reply drain (CI1)
+ * ════════════════════════════════════════════════════════════════════════ */
+
+int xcecli_ctx_recover(api_ctx_t* ac)
+{
+    if (ac == NULL)
+        return -EINVAL;
+
+    /* CI2 — state reset: the transport-error path (handle_disconnect) left
+     * ctx->state = XCE_CTX_ERR, which the per-op XCE_CTX_RUN guard in xceapi.c
+     * rejects as -EINVAL — the reported bug. After a successful reconnect,
+     * restore RUN so the next xce_* attempt passes. ONLY flip ERR→RUN: a
+     * deliberately-stopped (DOWN) or not-yet-started (INIT/BUSY) ctx is left
+     * untouched. */
+    if (ac->ctx.state == XCE_CTX_ERR)
+        ac->ctx.state = XCE_CTX_RUN;
+
+    /* CI1 — stale-reply drain: handle_disconnect may have injected a stale
+     * transport status into the single-slot wbuf. Drain any pending entry with
+     * a NON-BLOCKING get so the next attempt's cbuf_get cannot consume a
+     * prior-attempt reply. cbuf_tryget returns -EAGAIN when the channel is
+     * already empty — the normal, no-op case. The reconnect closed the old fd
+     * first, so no further replies arrive from the dead connection. */
+    int stale = 0;
+    cbuf_tryget(&ac->ctx.wbuf, &stale);
+
+    return 0;
+}
+
 /* ════════════════════════════════════════════════════════════════════════
  * XcecliRetryLoop — decision core (pure) + side-effecting wrapper
  * ════════════════════════════════════════════════════════════════════════ */
@@ -289,6 +319,11 @@ bool xcecli_retry_next(api_ctx_t* ac, int* rc, bool is_write,
     if (rrc < 0) {
         WARN("xcecli failover: rotate failed (rc=%d); retrying anyway", rrc);
     } else {
+        /* Reconnect succeeded: repair the ctx before the next xce_* attempt
+         * (CI2 ERR→RUN state reset, CI1 stale-reply drain) so the per-op guard
+         * passes and no prior-attempt reply is consumed (1.2, 4.4). */
+        xcecli_ctx_recover(ac);
+
         /* Log the rotation target (7.1). xce_failover_rotate consumes the
          * redirect hint when present (so the post-rotate target is not always
          * the sticky master); report the sticky master cursor as a best-effort

src/lib/client_internal.h

@@ -139,4 +139,25 @@ bool xcecli_retry_decide(api_ctx_t* ac, int* rc, bool is_write,
 bool xcecli_retry_next(api_ctx_t* ac, int* rc, bool is_write,
                        xcecli_retry_t* st);
 
+/*
+ * xcecli_ctx_recover (task 3.2) — repair the inner xce_ctx_t after a successful
+ * reconnect so the next xce_* attempt can proceed. Socket-free (no I/O, no
+ * sleep) so it is unit-testable; xcecli_retry_next invokes it after a successful
+ * xce_failover_rotate(), before re-sending the interrupted op.
+ *
+ *   CI2 (state reset): the transport-error path left ctx->state = XCE_CTX_ERR
+ *     (xcectx.c:174), which the per-op XCE_CTX_RUN guard rejects as -EINVAL
+ *     (xceapi.c) — the reported bug. recover restores XCE_CTX_RUN, but ONLY from
+ *     ERR: a DOWN/INIT ctx (deliberately stopped / not yet started) is left
+ *     as-is, and a RUN ctx is already fine.
+ *   CI1 (stale-reply drain): handle_disconnect may have injected a stale status
+ *     into the single-slot ctx->wbuf (xcectx.c:180). recover drains any pending
+ *     entry with a non-blocking cbuf_tryget so the next attempt's cbuf_get
+ *     cannot consume a prior-attempt reply. Reconnect closes the old fd first,
+ *     so the dead connection yields no further replies — this drain is enough.
+ *
+ * Returns 0 on success, -EINVAL on a NULL ac.
+ */
+int xcecli_ctx_recover(api_ctx_t* ac);
+
 #endif

src/lib/testsuite/test_libxce_retry.c

@@ -1021,6 +1021,112 @@ static void test_decide_transport_exhaustion_keeps_last_rc(void)
     printf("test_decide_transport_exhaustion_keeps_last_rc: PASS\n");
 }
 
+/* ════════════════════════════════════════════════════════════════════════
+ * Task 3.2 — Reconnect recovery (xcecli_ctx_recover) RED→GREEN tests.
+ *
+ * After a transport failure the disconnect path leaves ctx->state == XCE_CTX_ERR
+ * (xcectx.c:174) and may have injected a stale transport status into the single-
+ * slot ctx->wbuf (xcectx.c:180). xcecli_ctx_recover repairs the ctx after a
+ * successful reconnect, with NO socket I/O so it is unit-testable:
+ *   - CI2 state reset: ERR → RUN (only ERR; RUN/DOWN/INIT left as-is) so the
+ *     next xce_* op passes the xceapi XCE_CTX_RUN guard instead of -EINVAL.
+ *   - CI1 stale-reply drain: any entry already sitting in wbuf is drained with a
+ *     non-blocking get, so the next attempt's cbuf_get cannot consume a prior-
+ *     attempt reply.
+ *
+ * Requirements: 1.2 (complete next request against new primary, no caller
+ * change), 4.4 (transport → reconnect/rotate and resume).
+ * ════════════════════════════════════════════════════════════════════════ */
+
+/* CI2: an ERR-state ctx is restored to RUN (the reported -EINVAL bug fix). */
+static void test_recover_resets_err_state_to_run(void)
+{
+    api_ctx_t ac;
+    memset(&ac, 0, sizeof(ac));
+    ac.ctx.state = XCE_CTX_ERR;
+    /* wbuf left un-init: recover must tolerate a recover that has nothing to
+     * drain (init it so the drain path is also exercised safely). */
+    TEST_ASSERT_EQUAL_INT(0, cbuf_init(&ac.ctx.wbuf, 1, sizeof(int)));
+
+    int rc = xcecli_ctx_recover(&ac);
+    cbuf_fini(&ac.ctx.wbuf);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(XCE_CTX_RUN, ac.ctx.state);
+
+    printf("test_recover_resets_err_state_to_run: PASS\n");
+}
+
+/* CI2: a ctx already in RUN is left in RUN; only ERR→RUN is performed. */
+static void test_recover_leaves_run_state_untouched(void)
+{
+    api_ctx_t ac;
+    memset(&ac, 0, sizeof(ac));
+    ac.ctx.state = XCE_CTX_RUN;
+    TEST_ASSERT_EQUAL_INT(0, cbuf_init(&ac.ctx.wbuf, 1, sizeof(int)));
+
+    int rc = xcecli_ctx_recover(&ac);
+    cbuf_fini(&ac.ctx.wbuf);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(XCE_CTX_RUN, ac.ctx.state);
+
+    printf("test_recover_leaves_run_state_untouched: PASS\n");
+}
+
+/* CI2: a DOWN ctx is NOT flipped to RUN — only an ERR transport state is
+ * repaired; a deliberately-stopped ctx stays down. */
+static void test_recover_leaves_down_state_untouched(void)
+{
+    api_ctx_t ac;
+    memset(&ac, 0, sizeof(ac));
+    ac.ctx.state = XCE_CTX_DOWN;
+    TEST_ASSERT_EQUAL_INT(0, cbuf_init(&ac.ctx.wbuf, 1, sizeof(int)));
+
+    int rc = xcecli_ctx_recover(&ac);
+    cbuf_fini(&ac.ctx.wbuf);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(XCE_CTX_DOWN, ac.ctx.state);
+
+    printf("test_recover_leaves_down_state_untouched: PASS\n");
+}
+
+/* CI1: a stale status already sitting in wbuf is drained, so a subsequent
+ * non-blocking get finds the channel EMPTY (the prior-attempt reply was
+ * discarded, not handed to the next attempt's cbuf_get). */
+static void test_recover_drains_stale_wbuf_entry(void)
+{
+    api_ctx_t ac;
+    memset(&ac, 0, sizeof(ac));
+    ac.ctx.state = XCE_CTX_ERR;
+    TEST_ASSERT_EQUAL_INT(0, cbuf_init(&ac.ctx.wbuf, 1, sizeof(int)));
+
+    /* mirror handle_disconnect: a stale transport status sits in the channel */
+    int stale = -ECONNRESET;
+    TEST_ASSERT_EQUAL_INT(0, cbuf_tryput(&ac.ctx.wbuf, &stale));
+
+    int rc = xcecli_ctx_recover(&ac);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(XCE_CTX_RUN, ac.ctx.state);
+
+    /* the stale entry was drained: a non-blocking get now finds nothing */
+    int drained = 0;
+    int g       = cbuf_tryget(&ac.ctx.wbuf, &drained);
+    cbuf_fini(&ac.ctx.wbuf);
+
+    TEST_ASSERT_EQUAL_INT(-EAGAIN, g); /* channel empty after recover */
+
+    printf("test_recover_drains_stale_wbuf_entry: PASS\n");
+}
+
+/* Guard: a NULL api_ctx is rejected without crashing. */
+static void test_recover_null_ctx_safe(void)
+{
+    TEST_ASSERT_EQUAL_INT(-EINVAL, xcecli_ctx_recover(NULL));
+    printf("test_recover_null_ctx_safe: PASS\n");
+}
+
 /* ── Suite entry point ──────────────────────────────────────────────── */
 
 void test_libxce_retry(void)
@@ -1054,5 +1160,10 @@ void test_libxce_retry(void)
     RUN_TEST(test_decide_terminal_surfaces_unchanged);
     RUN_TEST(test_decide_dup_on_resent_write_is_success);
     RUN_TEST(test_decide_transport_exhaustion_keeps_last_rc);
+    RUN_TEST(test_recover_resets_err_state_to_run);
+    RUN_TEST(test_recover_leaves_run_state_untouched);
+    RUN_TEST(test_recover_leaves_down_state_untouched);
+    RUN_TEST(test_recover_drains_stale_wbuf_entry);
+    RUN_TEST(test_recover_null_ctx_safe);
     UNITY_END();
 }