#if !defined(_REQUEST_H_) #define _REQUEST_H_ 1 #include #include #include #include #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*); 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); extern int xce_notify_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*); /* * 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); /* ── 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