Rongsong Shen
0cad68507f
A worker reported job start, finish and failure with a fire-and-forget
OPCode_notify the controller never answers, so a report lost while the
worker stayed connected (the live incident: 6 of 7 finish reports in a
burst) left the job RUN forever; the rejoin resync (165fc158) only
recovers reports lost around a reconnect.
- controller.capnp: OPCode_notifyAcked -- the same NotifyData body,
answered. A separate opcode because replies are matched through the
client's single wbuf, not by sequence number: answering plain notifies
would feed stale statuses to the next synchronous request.
- controller (cluster.c): do_notify_acked() applies the fact like
do_notify, then waits on the commit barrier and replies 0 only once it
is majority-committed. A barrier failure (the fact is still appended and
retained) replies -E_XCE_NOT_READY, so the worker retries in place
instead of rotating away from a healthy primary (cf. bug-1850). Admitted
by the no-quorum fact-bearing gate like OPCode_notify.
- libxce: xce_notify_acked_req/xce_notify_acked (send, wait for the
status) and xcecli_notify_acked (XCECLI_FAILOVER_CALL retry/failover).
xcecli_notify_jobstart/jobfinish/jobfail use it; on -E_XCE_BAD_OPCODE
(a controller predating the opcode) they fall back to the plain
fire-and-forget notify, so workers and controllers can be upgraded in
either order.
- Resends are safe: START/FINISH are duplicate-guarded and a repeated FAIL
yields the same state.
- worker: logs a WARN when a report is still unacknowledged after the
retry budget (the rejoin resync remains the backstop). These calls run
on the worker daemon's IPC callback, so while the controller lacks
quorum other jobs' IPC messages wait (buffered, not lost) up to the
retry budget.
Tests: test_notify_acked_reply_only_after_commit and
test_notify_acked_withheld_replies_not_ready (no reply before the covering
ACK; -E_XCE_NOT_READY with the fact retained when ACKs are withheld).
Written with the opcode, so no separate RED run. Full /xce-build all:
111 suites pass; controller suite 1245/0.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
256 lines
12 KiB
C
256 lines
12 KiB
C
#if !defined(_REQUEST_H_)
|
|
|
|
#define _REQUEST_H_ 1
|
|
|
|
#include <stddef.h>
|
|
#include <stdint.h>
|
|
#include <stdlib.h>
|
|
|
|
#include <openssl/ssl.h>
|
|
|
|
#include "netio.h"
|
|
#include "typemap.h"
|
|
#include "utils.h"
|
|
#include "xce/types.h"
|
|
|
|
typedef struct {
|
|
int wait;
|
|
} reqbar_t;
|
|
|
|
typedef int (*xce_check_fn_t)(void*);
|
|
|
|
extern int xce_request(endpoint_t *ep, xce_request_t *req);
|
|
extern int xce_encode(xce_request_t*, io_packet_t*);
|
|
extern int xce_decode(io_packet_t*, xce_request_t*);
|
|
|
|
extern int xce_reply(endpoint_t*, xce_peerinfo_t*, xce_reply_t*);
|
|
extern int xce_hello_req(endpoint_t*, char*);
|
|
extern int xce_nop_req(endpoint_t*);
|
|
/* HA heartbeat: an OP_nop request carrying a HeartbeatMsg body with the
|
|
* primary's committed position and its own election time (Req 12.2 — echoed
|
|
* to standbys so they can learn the currently-known primary's election
|
|
* time). Non-blocking, no reply awaited (the standby's StateReport rides the
|
|
* normal nop reply). senderPriority is unused (0). Generic xce_nop_req is
|
|
* left unchanged (body-less). */
|
|
extern int xce_ha_heartbeat_req(endpoint_t* ep, uint64_t term,
|
|
uint64_t committed_lsn,
|
|
uint64_t committed_config_tag,
|
|
uint64_t elected_at_ms);
|
|
extern int xce_shutdown_req(endpoint_t*);
|
|
extern void init_reply_by_req(xce_reply_t*, xce_request_t*);
|
|
extern int xce_join_req(endpoint_t*, int, hostattrs_t*);
|
|
extern int xce_pushconfig_req(endpoint_t*, xce_peerinfo_t*, xce_configdata_t*);
|
|
extern int xce_push_data_req(endpoint_t*, xce_peerinfo_t*, xce_data_t*,
|
|
xce_check_fn_t, void*, int);
|
|
|
|
extern int xce_workerlist_req(endpoint_t*, xce_peerinfo_t*);
|
|
extern int xce_reply_workerlist(endpoint_t*, uint64_t, int, xce_peerinfo_t*,
|
|
xce_workerlist_t*);
|
|
extern int xce_jobdispatch_req(endpoint_t*, xce_job_dispatch_t*);
|
|
extern int xce_jobsubmit_req(endpoint_t*, xce_job_submit_t*);
|
|
extern int xce_reply_jobdispatch(endpoint_t*, uint64_t, int32_t,
|
|
xce_peerinfo_t*);
|
|
extern int xce_reply_jobid(endpoint_t*, int, uint64_t, uint64_t,
|
|
xce_peerinfo_t*);
|
|
extern int xce_joblist_req(endpoint_t*, xce_peerinfo_t*);
|
|
extern int xce_reply_joblist(endpoint_t*, uint64_t, int, xce_peerinfo_t*,
|
|
xce_job_list_t*);
|
|
extern int xce_startjob_req(endpoint_t*, xce_job_exec_t*);
|
|
extern int xce_clusterconf_req(endpoint_t*);
|
|
extern int xce_reply_clusterconf(endpoint_t*, uint64_t, int, xce_peerinfo_t*,
|
|
xce_clusterconf_t*);
|
|
extern int xce_ctrlstatus_req(endpoint_t*);
|
|
extern int xce_get_req(endpoint_t*, xce_get_t*);
|
|
extern int xce_reply_get(endpoint_t*, xce_peerinfo_t*, xce_reply_t*,
|
|
xce_check_fn_t, void*, int);
|
|
|
|
/*
|
|
* xce_job_state_query_req / xce_job_state_report_req: send the worker-rejoin-
|
|
* resync job-state reconciliation exchange's paginated request/reply pair
|
|
* (worker-rejoin-resync task 1.4, Req 8.1/8.2/8.3) --
|
|
* OPCode_jobStateQuery (controller -> worker, query->jobids) and
|
|
* OPCode_jobStateReport (worker -> controller, report->entries) -- using the
|
|
* same {pos,count,ntotal} pagination idiom as xce_push_data_req()/
|
|
* xce_reply_get() above: `pktlimit <= 0` sends the whole list in one page;
|
|
* a zero-length list still sends exactly one page with count == 0 (an empty
|
|
* page is a valid, complete exchange, not something to skip). The receiving
|
|
* side decodes each page with the generic xce_decode() -- these opcodes are
|
|
* their own dedicated Request body union arms (jobstatequerymsg/
|
|
* jobstatereportmsg, added by task 1.1's schema), not multiplexed through
|
|
* xce_data_t, so encode/decode of each page's payload is already handled by
|
|
* the generated encode_JobStateQuery()/encode_JobStateReport() (controller.capnp.c)
|
|
* that xce_encode()/xce_decode() call into.
|
|
*/
|
|
extern int xce_job_state_query_req(endpoint_t*, xce_peerinfo_t*,
|
|
xce_job_state_query_t*,
|
|
xce_check_fn_t, void*, int);
|
|
extern int xce_job_state_report_req(endpoint_t*, xce_peerinfo_t*,
|
|
xce_job_state_report_t*,
|
|
xce_check_fn_t, void*, int);
|
|
|
|
extern int xce_notify_req(endpoint_t*, xce_notify_data_t*);
|
|
/* bug-1741: OPCode_notifyAcked -- answered once the fact commits. */
|
|
extern int xce_notify_acked_req(endpoint_t*, xce_notify_data_t*);
|
|
extern int extract_data(dataset_t*, xce_data_t*);
|
|
extern int xce_update_req(endpoint_t*, xce_update_data_t*);
|
|
extern int xce_jobctrl_req(endpoint_t*, xce_jobctrl_t*);
|
|
extern int xce_queuectrl_req(endpoint_t*, xce_queuectrl_t*);
|
|
extern int xce_hostctrl_req(endpoint_t*, xce_hostctrl_t*);
|
|
extern int xce_runtask_req(endpoint_t*, xce_taskrun_t*);
|
|
extern int xce_set_pend_reason_req(endpoint_t*, xce_preason_set_t*);
|
|
extern int xce_clear_reason_req(endpoint_t*, xce_clear_reason_t*);
|
|
extern int xce_fetchfile_req(endpoint_t*, xce_fetchfile_t*);
|
|
|
|
/*
|
|
* xce_schedadmin_req: send OPCode_schedAdmin (SchedAdminRequest) to `ep` --
|
|
* the async send half of task 10.1's controller->scheduler forward. Mirrors
|
|
* xce_fetchfile_req above exactly. The caller (myreq_handler, cluster.c)
|
|
* blocks separately for the matching SchedAdminReply; this function only
|
|
* enqueues the request, it does not wait.
|
|
*/
|
|
extern int xce_schedadmin_req(endpoint_t*, sched_admin_req_t*);
|
|
|
|
/*
|
|
* xce_replconfigpush_req_raw: send OP_replConfigPush carrying raw byte arrays.
|
|
* Bypasses the encode_ReplConfigPushRequest_ptr path (which maps sha256/bytes
|
|
* through xce_data_t — wrong type for raw bytes) by constructing the capnp
|
|
* message directly via capn_new_list8 + capn_setv8 + write_Request.
|
|
*/
|
|
extern int xce_replconfigpush_req_raw(endpoint_t *ep, xce_peerinfo_t *from,
|
|
uint64_t version_tag,
|
|
const uint8_t *sha256, size_t sha256_len,
|
|
const uint8_t *config_bytes,
|
|
size_t config_len);
|
|
|
|
/*
|
|
* xce_replconfig_reply_raw: primary sends an OP_replConfig reply with
|
|
* actual config bytes via endpoint (iomgr queued). Bypasses
|
|
* encode_ReplConfigReply_ptr (which maps sha256/bytes through xce_data_t).
|
|
*/
|
|
extern int xce_replconfig_reply_raw(endpoint_t *ep, xce_peerinfo_t *from,
|
|
uint64_t seq, uint64_t tag,
|
|
const uint8_t *sha256, size_t sha256_len,
|
|
const uint8_t *config, size_t config_len);
|
|
|
|
/*
|
|
* xce_replconfig_req_ssl: standby sends OP_replConfig over a raw SSL
|
|
* socket (no iomgr). Wire framing: uint64_t size + capnp payload
|
|
* (matches netio.c io_fsm_perform).
|
|
*/
|
|
extern int xce_replconfig_req_ssl(SSL *ssl, uint64_t version_tag);
|
|
|
|
/*
|
|
* xce_replconfig_reply_ssl_decode: standby reads + parses an
|
|
* OP_replConfig reply from a raw SSL socket. Output buffers are
|
|
* malloc'd and owned by the caller. Both *out_sha256 and *out_bytes
|
|
* may be NULL on the up_to_date == 1 fast path.
|
|
*/
|
|
extern int xce_replconfig_reply_ssl_decode(SSL *ssl,
|
|
int *out_up_to_date,
|
|
uint64_t *out_tag,
|
|
uint8_t **out_sha256,
|
|
uint8_t **out_bytes,
|
|
size_t *out_len);
|
|
|
|
/*
|
|
* xce_free_request: free an xce_request_t and all owned sub-fields,
|
|
* including the raw_sha256/raw_bytes fields on upgrade_config_req_t
|
|
* that the generated free_Request() does not know about.
|
|
* Use this everywhere instead of calling free_Request() directly.
|
|
*/
|
|
extern void xce_free_request(xce_request_t *req);
|
|
|
|
/*
|
|
* xce_wal_batch_encode_raw / xce_wal_batch_req_raw: encode/send an
|
|
* OPCode_walCatchupEntry batch carrying raw WAL payload bytes.
|
|
*
|
|
* Bypasses the generated encode path (payload :Data cannot be carried
|
|
* through xce_data_t — same limitation as replConfigPush). Each
|
|
* entries[i]->raw_payload/raw_payload_len is written as a capnp Data
|
|
* (List(UInt8)); the legacy xce_data_t payload field is ignored.
|
|
* The receiver's xce_decode() extracts the bytes back into
|
|
* raw_payload/raw_payload_len on each decoded sub-entry.
|
|
*
|
|
* encode_raw fills pkt (caller frees via free_packet on error paths);
|
|
* req_raw allocates the packet and enqueues it on ep (non-blocking).
|
|
*/
|
|
extern int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries,
|
|
int n_entries,
|
|
xce_peerinfo_t *from,
|
|
uint64_t seq,
|
|
io_packet_t *pkt);
|
|
extern int xce_wal_batch_req_raw(endpoint_t *ep,
|
|
wal_catchup_entry_t **entries,
|
|
int n_entries);
|
|
|
|
/*
|
|
* xce_snapshot_chunk_encode_raw / xce_snapshot_chunk_req_raw: encode/send
|
|
* an OPCode_snapshotTransfer chunk carrying raw snapshot file bytes.
|
|
* Same raw-Data convention as xce_wal_batch_req_raw; chunk envelope
|
|
* fields (snapshot_lsn, chunk_index, chunk_offset, total_size, is_last)
|
|
* are taken from *chunk, payload bytes from payload/payload_len.
|
|
*/
|
|
extern int xce_snapshot_chunk_encode_raw(const snapshot_transfer_t *chunk,
|
|
const uint8_t *payload,
|
|
size_t payload_len,
|
|
xce_peerinfo_t *from,
|
|
uint64_t seq,
|
|
io_packet_t *pkt);
|
|
extern int xce_snapshot_chunk_req_raw(endpoint_t *ep,
|
|
const snapshot_transfer_t *chunk,
|
|
const uint8_t *payload,
|
|
size_t payload_len);
|
|
|
|
/*
|
|
* xce_upgrade_config_req_extract: extract sha256 + bytes from a decoded
|
|
* OP_upgradeConfig request. xce_decode() stores malloc'd copies in
|
|
* req->body.upgradeconfigrequestmsg->raw_sha256 / raw_bytes before
|
|
* capn_free() runs. Returns shallow pointers (owned by req); caller
|
|
* must NOT free them. Returns -EINVAL if fields are missing.
|
|
*/
|
|
extern int xce_upgrade_config_req_extract(const xce_request_t *req,
|
|
uint8_t **out_bytes,
|
|
size_t *out_len,
|
|
uint8_t **out_sha256);
|
|
|
|
/*
|
|
* xce_upgrade_config_req_extract_raw: parse OP_upgradeConfig request
|
|
* bytes from a raw on-wire packet buffer. Output buffers (*out_bytes,
|
|
* *out_sha256) are malloc'd and owned by the caller.
|
|
*/
|
|
extern int xce_upgrade_config_req_extract_raw(const uint8_t *pkt_data,
|
|
size_t pkt_sz,
|
|
uint8_t **out_bytes,
|
|
size_t *out_len,
|
|
uint8_t **out_sha256);
|
|
|
|
/*
|
|
* xce_upgradeconfig_req_raw: send OP_upgradeConfig carrying raw byte arrays
|
|
* (sha256 + the serialised ClusterConfig blob). Bypasses the generated
|
|
* encode_UpgradeConfigRequest path (a no-op for the sha256/bytes Data fields
|
|
* — same class of limitation as xce_replconfigpush_req_raw) by constructing
|
|
* the capnp message directly via capn_new_list8 + capn_setv8 + write_Request.
|
|
* Ordinary CLIENT request: seq/from are populated the same way xce_request()
|
|
* populates them for every other client op.
|
|
*/
|
|
extern int xce_upgradeconfig_req_raw(endpoint_t *ep,
|
|
const uint8_t *sha256, size_t sha256_len,
|
|
const uint8_t *bytes, size_t bytes_len);
|
|
|
|
/* ── Client-Side Transparent Retry tunables (Req 6.1, 6.2) ──────────────────
|
|
*
|
|
* xce_retry_config_t: configurable retry parameters with safe production
|
|
* defaults, consumed by the libxce client failover loop (client.c /
|
|
* client_failover.c). XCE_RETRY_DEFAULTS supplies the defaults applied when no
|
|
* override is configured.
|
|
*/
|
|
typedef struct {
|
|
int max_retries; /* default 20 */
|
|
int initial_retry_ms; /* default 50 */
|
|
int max_retry_ms; /* default 5000 */
|
|
double backoff_multiplier; /* default 2.0 (exponential) */
|
|
} xce_retry_config_t;
|
|
|
|
extern const xce_retry_config_t XCE_RETRY_DEFAULTS;
|
|
|
|
#endif
|