From 42279431f3c2384e3dda7422383c119f8dde9d9f Mon Sep 17 00:00:00 2001 From: Cococry Date: Mon, 27 Jul 2026 23:29:45 +0200 Subject: [PATCH] feat: Event layer on top of envelope protocol --- faithd/nob.c | 8 +- faithd/src/client/client.c | 191 ++++++++++++++++++----------- faithd/src/codec/commands.h | 3 +- faithd/src/codec/envelopes.h | 28 +---- faithd/src/codec/events.c | 43 +++++++ faithd/src/codec/events.h | 63 ++++++++++ faithd/src/codec/protocol.c | 31 ++++- faithd/src/codec/protocol.h | 11 ++ faithd/src/commands/conversation.c | 48 ++++++-- faithd/src/core/core.h | 3 +- faithd/src/delivery/event_inbox.c | 62 ++++++++++ faithd/src/delivery/event_inbox.h | 25 ++++ faithd/src/delivery/events.c | 75 +++++++++++ faithd/src/delivery/events.h | 14 +++ faithd/src/server/dispatch.c | 67 +++++----- faithd/src/server/dispatch.h | 6 - faithd/src/server/sess_registry.c | 7 +- faithd/src/server/sess_registry.h | 2 + faithd/third_party/stb_ds.h | 28 +++-- 19 files changed, 558 insertions(+), 157 deletions(-) create mode 100644 faithd/src/codec/events.c create mode 100644 faithd/src/codec/events.h create mode 100644 faithd/src/delivery/event_inbox.c create mode 100644 faithd/src/delivery/event_inbox.h create mode 100644 faithd/src/delivery/events.c create mode 100644 faithd/src/delivery/events.h diff --git a/faithd/nob.c b/faithd/nob.c index 5069e8c..d765cc1 100644 --- a/faithd/nob.c +++ b/faithd/nob.c @@ -84,6 +84,7 @@ static const char *faithd_sources[] = { "src/auth/device_link.c", "src/auth/handshake.c", "src/auth/structs.c", + "src/codec/events.c", "src/codec/helpers.c", "src/codec/msg.c", "src/codec/protocol.c", @@ -92,6 +93,8 @@ static const char *faithd_sources[] = { "src/core/core.c", "src/core/crypto.c", "src/codec/envelopes.c", + "src/delivery/event_inbox.c", + "src/delivery/events.c", "src/delivery/routing.c", "src/logging/logging.c", "src/reactor/reactor.c", @@ -112,6 +115,7 @@ static const char *faithd_objects[] = { BUILD_FOLDER "obj/auth_device_link.o", BUILD_FOLDER "obj/auth_handshake.o", BUILD_FOLDER "obj/auth_structs.o", + BUILD_FOLDER "obj/codec_events.o", BUILD_FOLDER "obj/codec_helpers.o", BUILD_FOLDER "obj/codec_msg.o", BUILD_FOLDER "obj/codec_protocol.o", @@ -120,6 +124,8 @@ static const char *faithd_objects[] = { BUILD_FOLDER "obj/core_core.o", BUILD_FOLDER "obj/core_crypto.o", BUILD_FOLDER "obj/codec_envelopes.o", + BUILD_FOLDER "obj/delivery_event_inbox.o", + BUILD_FOLDER "obj/delivery_events.o", BUILD_FOLDER "obj/delivery_routing.o", BUILD_FOLDER "obj/logging_logging.o", BUILD_FOLDER "obj/reactor_reactor.o", @@ -151,7 +157,7 @@ static const char *test_shared_objects[] = { BUILD_FOLDER "obj/core_core.o", BUILD_FOLDER "obj/codec_helpers.o", BUILD_FOLDER "obj/codec_envelopes.o", BUILD_FOLDER "obj/auth_structs.o", BUILD_FOLDER "obj/codec_signatures.o", BUILD_FOLDER "obj/codec_msg.o", - BUILD_FOLDER "obj/codec_commands.o", + BUILD_FOLDER "obj/codec_commands.o", BUILD_FOLDER "obj/codec_events.o", }; static bool compile_source_async(Procs *procs, const char *source, diff --git a/faithd/src/client/client.c b/faithd/src/client/client.c index 1c87338..020dba3 100644 --- a/faithd/src/client/client.c +++ b/faithd/src/client/client.c @@ -123,6 +123,8 @@ struct faith_client { pthread_cond_t reconnect_cond; pending_command_entry_t *pending_command_map; + + uint64_t last_acked_ev_seq; }; static faith_status_code_t @@ -162,7 +164,7 @@ static faith_status_code_t pending_command_take(faith_client_t *client, if (!cmd) _FH_RETURN_DEFER(FAITH_ERR_NOT_FOUND); - hmdel(client->pending_command_map, *id); + (void)hmdel(client->pending_command_map, *id); defer: pthread_mutex_unlock(&client->command_lock); @@ -280,7 +282,7 @@ static faith_status_code_t read_frame_sync(SSL *ssl, faith_frame_t *out) { read_rc == FAITH_ERR_IO) return read_rc; else if (read_rc == FAITH_ERR_CLOSED) { - return FAITH_OK; + return FAITH_ERR_CLOSED; } const uint32_t frame_size = faith_read_u32_be(len_buf); @@ -926,6 +928,102 @@ client_handle_disconnect(faith_client_t *client, const faith_envelope_t *envl) { return FAITH_OK; } + +static faith_status_code_t +client_send_ack_event(faith_client_t *client, uint64_t seq_num, + faith_event_codec_type_t event_type) { + + if(!client || seq_num == UINT64_MAX) return FAITH_ERR_INVALID; + + uint8_t body[FAITH_ENVL_CTS_EVENT_ACK_BODY_SIZE] = {0}; + faith_body_size_t body_size = 0; + + faith_envl_cts_event_ack_t ack = {0}; + ack.seq_num = seq_num; + ack.type = event_type; + + _FH_CHECK_RETURN( + faith_encode_event_ack_body(body, &body_size, sizeof(body), &ack)); + + faith_envelope_t envl = {0}; + envl.type = FAITH_ENVELOPE_EVENT_ACK; + envl.body = body; + envl.body_size = body_size; + + _FH_CHECK_RETURN(client_send_envelope_locked(client, &envl)); + + return FAITH_OK; +} + +static faith_status_code_t client_handle_event_conversation_created(faith_client_t* client, const faith_envl_stc_event_t* event) { + if(!client || !event) return FAITH_ERR_INVALID; + + faith_event_conversation_created_t conv_created = {0}; + _FH_CHECK_RETURN(faith_decode_event_conversation_created( + event->data, event->data_size, &conv_created)); + + char buf[33]; + _FH_CHECK_RETURN(faith_id128_to_hex(conv_created.conversation_id.bytes, buf)); + nob_log(INFO, "[client] Conversation %s was created by the server.", buf); + + return FAITH_OK; +} +static faith_status_code_t +client_dispatch_event(faith_client_t* client, const faith_envl_stc_event_t* event) { + if(!client || !event) return FAITH_ERR_INVALID; + + switch(event->type) { + case FAITH_EVENT_CONVERSATION_CREATED: + _FH_CHECK_RETURN(client_handle_event_conversation_created(client, event)); + break; + default: + break; + } + + return FAITH_OK; +} + +static faith_status_code_t client_ack_event(faith_client_t* client, const faith_envl_stc_event_t* event) { + if(!client || !event) return FAITH_ERR_INVALID; + + /* duplicate event */ + if (client->last_acked_ev_seq != UINT64_MAX) { + if (event->seq_num <= client->last_acked_ev_seq) + return FAITH_OK; + } + + /* TODO: Handle out of order events */ + + _FH_CHECK_RETURN(client_send_ack_event(client, event->seq_num, event->type)); + + client->last_acked_ev_seq = event->seq_num; + + return FAITH_OK; +} + +static faith_status_code_t +client_handle_event(faith_client_t *client, + const faith_envelope_t *envl) { + if (!client || !envl) + return FAITH_ERR_INVALID; + + faith_envl_stc_event_t event = {0}; + + if (event.seq_num == UINT64_MAX) { + nob_log(ERROR, "[client] Received event (%s) with invalid sequence number.", + faith_event_codec_type_name(event.type)); + return FAITH_ERR_INVALID; + } + + _FH_CHECK_RETURN( + faith_decode_event_body(envl->body, envl->body_size, &event)); + + _FH_CHECK_RETURN(client_dispatch_event(client, &event)); + + _FH_CHECK_RETURN(client_ack_event(client, &event)); + + return FAITH_OK; +} static faith_status_code_t client_handle_command_result(faith_client_t *client, const faith_envelope_t *envl) { @@ -1153,6 +1251,10 @@ static faith_status_code_t client_handle_envelope(faith_client_t *client, break; case FAITH_ENVELOPE_COMMAND_RESULT: _FH_CHECK_RETURN(client_handle_command_result(client, &envl)); + break; + case FAITH_ENVELOPE_EVENT: + _FH_CHECK_RETURN(client_handle_event(client, &envl)); + break; default: break; } @@ -1168,8 +1270,6 @@ static void *reader(void *arg) { faith_frame_t frame; if (read_frame_sync(client->ssl, &frame) != FAITH_OK) { - _FH_CHECK(client_push_event(client, FAITH_EVENT_DISCONNECTED, 0, 0, - "Failed to read server frame")); break; } @@ -1179,7 +1279,7 @@ static void *reader(void *arg) { break; } case FAITH_MSG_ENVL: { - client_handle_envelope(client, &frame); + _FH_CHECK(client_handle_envelope(client, &frame)); break; } default: @@ -1280,8 +1380,6 @@ static faith_status_code_t faith_client_send_hello(faith_client_t *client) { if (!client) return FAITH_ERR_INVALID; - uint8_t body[FAITH_ENVL_CTS_HELLO_BODY_SIZE]; - // HELLO { // header: { // sender_id: auth_id @@ -1294,32 +1392,24 @@ static faith_status_code_t faith_client_send_hello(faith_client_t *client) { // } // - size_t offset = 0; - size_t device_id_size = sizeof(client->ident.device_id.bytes); - /* 1. Serialize device ID */ - memcpy(body, client->ident.device_id.bytes, device_id_size); - offset += device_id_size; - - /* 2. Serialize public key*/ - memcpy(body + offset, client->ident.public_key, - sizeof(client->ident.public_key)); - - offset += sizeof(client->ident.public_key); + faith_envl_cts_hello_t hello = {0}; + hello.device_id = client->ident.device_id; + memcpy(hello.public_key, client->ident.public_key, + FAITH_ED25519_PUBLIC_KEY_SIZE); + _FH_CHECK_RETURN(faith_random_bytes((uint8_t *)&hello.client_nonce, + sizeof(hello.client_nonce))); - /* 3. Serialize nonce */ - uint64_t nonce; - _FH_CHECK_RETURN(faith_random_bytes((uint8_t *)&nonce, sizeof(nonce))); + uint8_t body[FAITH_ENVL_CTS_HELLO_BODY_SIZE] = {0}; + faith_body_size_t body_size = 0; - _FH_CHECK_RETURN(faith_write_u64_be(body + offset, nonce)); + _FH_CHECK_RETURN(faith_encode_hello_body(body, &body_size, sizeof(body), &hello)); /* Store nonce in temporary client state */ - client->nonce_tmp = nonce; + client->nonce_tmp = hello.client_nonce; faith_envelope_t envl = {0}; envl.type = FAITH_ENVELOPE_HELLO; - - memcpy(envl.sender_id.bytes, client->ident.auth_id.bytes, - sizeof(envl.sender_id)); + envl.sender_id = client->ident.auth_id; envl.body = body; envl.body_size = sizeof(body); @@ -1670,53 +1760,6 @@ faith_status_code_t faith_client_init_global(int log_enable_tracing) { return FAITH_OK; } -static bool hex_char_to_nibble(char c, uint8_t *out) { - if (!out) - return false; - - if (c >= '0' && c <= '9') { - *out = (uint8_t)(c - '0'); - return true; - } - - if (c >= 'a' && c <= 'f') { - *out = (uint8_t)(c - 'a' + 10); - return true; - } - - if (c >= 'A' && c <= 'F') { - *out = (uint8_t)(c - 'A' + 10); - return true; - } - - return false; -} - -static bool client_id_from_hex(const char *hex, faith_auth_id_t *out) { - if (!hex || !out) - return false; - - if (strlen(hex) != FAITH_AUTH_ID_SIZE * 2) - return false; - - faith_auth_id_t id = {0}; - - for (size_t i = 0; i < FAITH_AUTH_ID_SIZE; ++i) { - uint8_t hi = 0; - uint8_t lo = 0; - - if (!hex_char_to_nibble(hex[i * 2 + 0], &hi)) - return false; - - if (!hex_char_to_nibble(hex[i * 2 + 1], &lo)) - return false; - - id.bytes[i] = (uint8_t)((hi << 4) | lo); - } - - *out = id; - return true; -} static faith_status_code_t client_new_identity(client_side_identity_t *o_ident) { @@ -1764,6 +1807,8 @@ faith_client_t *faith_client_create(const faith_client_config_t *cfg) { client->sockfd = -1; client->event_fd = -1; + client->last_acked_ev_seq = UINT64_MAX; + snprintf(client->host, sizeof(client->host), "%s", cfg->host); client->port = cfg->port; diff --git a/faithd/src/codec/commands.h b/faithd/src/codec/commands.h index 11c1c01..d9c0185 100644 --- a/faithd/src/codec/commands.h +++ b/faithd/src/codec/commands.h @@ -30,7 +30,8 @@ typedef enum { X(FAITH_COMMAND_ERR_NONE, 0) \ X(FAITH_COMMAND_ERR_UNAUTHORIZED, 1) \ X(FAITH_COMMAND_ERR_BAD_COMMAND, 2) \ - X(FAITH_COMMAND_ERR_TIMED_OUT, 3) + X(FAITH_COMMAND_ERR_TIMED_OUT, 3) \ + X(FAITH_COMMAND_ERR_INTERNAL_ERROR, 4) typedef enum { #define X(name, value) name = value, diff --git a/faithd/src/codec/envelopes.h b/faithd/src/codec/envelopes.h index 2343dae..7714941 100644 --- a/faithd/src/codec/envelopes.h +++ b/faithd/src/codec/envelopes.h @@ -21,7 +21,8 @@ X(FAITH_ENVELOPE_CLIENT_DISCONNECT, 15) \ X(FAITH_ENVELOPE_COMMAND, 16) \ X(FAITH_ENVELOPE_COMMAND_RESULT, 17) \ - X(FAITH_ENVELOPE_EVENT, 18) + X(FAITH_ENVELOPE_EVENT, 18) \ + X(FAITH_ENVELOPE_EVENT_ACK, 19) typedef enum { #define X(name, value) name = value, @@ -29,20 +30,6 @@ typedef enum { #undef X } faith_envelope_type_t; -#define FAITH_ENVL_STC_EVENT_BODY_SIZE_FIXED \ - _FAITH_BODY_SIZE(sizeof(uint64_t) /* sequence number */ + \ - sizeof(uint32_t) /* type */ + \ - sizeof(faith_body_size_t) /* data size */ \ - ) - -#define FAITH_EVENTS_ENVL(X) X(FAITH_EVENT_CONVERSATION_CREATED, 0) - -typedef enum { -#define X(name, value) name = value, - FAITH_EVENTS_ENVL(X) -#undef X -} faith_event_envelope_type_t; - typedef struct { faith_envelope_type_t type; @@ -53,15 +40,4 @@ typedef struct { uint8_t *body; } faith_envelope_t; -typedef struct { - /* The server-generated sequence number - * of this event */ - uint64_t seq_num; - - faith_event_envelope_type_t type; - - faith_body_size_t data_size; - uint8_t *data; -} faith_envl_stc_event_t; - const char *faith_envelope_name(faith_envelope_type_t env); diff --git a/faithd/src/codec/events.c b/faithd/src/codec/events.c new file mode 100644 index 0000000..97298b0 --- /dev/null +++ b/faithd/src/codec/events.c @@ -0,0 +1,43 @@ +#include "events.h" + +faith_status_code_t faith_encode_event_conversation_created( + uint8_t *out_buf, faith_body_size_t *out_size, size_t buf_cap_in_bytes, + const faith_event_conversation_created_t *in) { + FAITH_ENCODE_PROLOGUE(FAITH_EVENT_CONVERSATION_CREATED_DATA_SIZE); + + size_t offset = 0; + + FAITH_APPEND_RETURN(out_buf, buf_cap_in_bytes, offset, + in->conversation_id.bytes, + sizeof(in->conversation_id.bytes)); + + FAITH_ENCODE_EPILOGUE(FAITH_EVENT_CONVERSATION_CREATED_DATA_SIZE, !=); + + return FAITH_OK; +} + +faith_status_code_t faith_decode_event_conversation_created( + const uint8_t *payload, faith_body_size_t payload_size, + faith_event_conversation_created_t *out) { + FAITH_DECODE_PROLOGUE(FAITH_EVENT_CONVERSATION_CREATED_DATA_SIZE, !=); + + size_t offset = 0; + + FAITH_DECODE_RETURN(payload, payload_size, offset, out->conversation_id.bytes, + sizeof(out->conversation_id)); + + FAITH_DECODE_EPILOGUE(FAITH_EVENT_CONVERSATION_CREATED_DATA_SIZE, !=); + + return FAITH_OK; +} +const char *faith_event_codec_type_name(faith_event_codec_type_t type) { + switch (type) { +#define X(name, value) \ + case name: \ + return #name; + FAITH_EVENTS_CODEC(X) +#undef X + default: + return "FAITH_EVENT_UNKNOWN"; + } +} diff --git a/faithd/src/codec/events.h b/faithd/src/codec/events.h new file mode 100644 index 0000000..25220ad --- /dev/null +++ b/faithd/src/codec/events.h @@ -0,0 +1,63 @@ +#pragma once + +#include "../auth/structs.h" +#include "core.h" +#include "helpers.h" + +#define FAITH_ENVL_STC_EVENT_BODY_SIZE_FIXED \ + _FAITH_BODY_SIZE(sizeof(uint64_t) /* sequence number */ + \ + sizeof(uint32_t) /* type */ + \ + sizeof(faith_body_size_t) /* data size */ \ + ) + +#define FAITH_ENVL_CTS_EVENT_ACK_BODY_SIZE \ + _FAITH_BODY_SIZE(sizeof(uint64_t) /* sequence number */) + +#define FAITH_EVENTS_CODEC(X) X(FAITH_EVENT_CONVERSATION_CREATED, 0) + +typedef enum { +#define X(name, value) name = value, + FAITH_EVENTS_CODEC(X) +#undef X +} faith_event_codec_type_t; + +typedef struct { + /* The server-generated sequence number + * of this event */ + uint64_t seq_num; + + faith_event_codec_type_t type; + + faith_body_size_t data_size; + uint8_t *data; +} faith_envl_stc_event_t; + +typedef struct { + /* The server-generated sequence number + * of the event that was acknowledged */ + uint64_t seq_num; + + faith_event_codec_type_t type; +} faith_envl_cts_event_ack_t; + +#define FAITH_CONVERSATION_ID_SIZE 16 + +typedef struct { + uint8_t bytes[FAITH_CONVERSATION_ID_SIZE]; +} faith_conversation_id_t; + +typedef struct { + faith_conversation_id_t conversation_id; +} faith_event_conversation_created_t; + +#define FAITH_EVENT_CONVERSATION_CREATED_DATA_SIZE _FAITH_BODY_SIZE(FAITH_CONVERSATION_ID_SIZE) + +faith_status_code_t faith_encode_event_conversation_created( + uint8_t *out_buf, faith_body_size_t *out_size, size_t buf_cap_in_bytes, + const faith_event_conversation_created_t *in); + +faith_status_code_t faith_decode_event_conversation_created( + const uint8_t *payload, faith_body_size_t payload_size, + faith_event_conversation_created_t *out); + +const char *faith_event_codec_type_name(faith_event_codec_type_t type); diff --git a/faithd/src/codec/protocol.c b/faithd/src/codec/protocol.c index b3ac253..059ba85 100644 --- a/faithd/src/codec/protocol.c +++ b/faithd/src/codec/protocol.c @@ -698,7 +698,7 @@ faith_status_code_t faith_encode_event_body(uint8_t *out_buf, in->data_size); } - FAITH_ENCODE_EPILOGUE(FAITH_ENVL_STC_EVENT_BODY_SIZE_FIXED + in->data, !=); + FAITH_ENCODE_EPILOGUE(FAITH_ENVL_STC_EVENT_BODY_SIZE_FIXED + in->data_size, !=); return FAITH_OK; } @@ -725,3 +725,32 @@ faith_status_code_t faith_decode_event_body(const uint8_t *payload, return FAITH_OK; } + +faith_status_code_t +faith_encode_event_ack_body(uint8_t *out_buf, faith_body_size_t *out_size, + size_t buf_cap_in_bytes, + const faith_envl_cts_event_ack_t *in) { + FAITH_ENCODE_PROLOGUE(FAITH_ENVL_CTS_EVENT_ACK_BODY_SIZE); + size_t offset = 0; + + FAITH_ENCODE_U64_BE_RETURN(out_buf, buf_cap_in_bytes, offset, in->seq_num); + FAITH_ENCODE_EPILOGUE(FAITH_ENVL_CTS_EVENT_ACK_BODY_SIZE, !=); + + return FAITH_OK; +} + +faith_status_code_t +faith_decode_event_ack_body(const uint8_t *payload, + faith_body_size_t payload_size, + faith_envl_cts_event_ack_t *out) { + + FAITH_DECODE_PROLOGUE(FAITH_ENVL_CTS_EVENT_ACK_BODY_SIZE, !=); + + size_t offset = 0; + + FAITH_DECODE_U64_BE_RETURN(payload, payload_size, offset, out->seq_num); + + FAITH_DECODE_EPILOGUE(FAITH_ENVL_CTS_EVENT_ACK_BODY_SIZE, !=); + + return FAITH_OK; +} diff --git a/faithd/src/codec/protocol.h b/faithd/src/codec/protocol.h index 4d44013..2165951 100644 --- a/faithd/src/codec/protocol.h +++ b/faithd/src/codec/protocol.h @@ -8,6 +8,7 @@ #include "../server/envelopes.h" #include "commands.h" +#include "events.h" #include "helpers.h" @@ -148,3 +149,13 @@ faith_status_code_t faith_encode_event_body(uint8_t *out_buf, faith_status_code_t faith_decode_event_body(const uint8_t *payload, faith_body_size_t payload_size, faith_envl_stc_event_t *out); + +faith_status_code_t +faith_encode_event_ack_body(uint8_t *out_buf, faith_body_size_t *out_size, + size_t buf_cap_in_bytes, + const faith_envl_cts_event_ack_t *in); + +faith_status_code_t +faith_decode_event_ack_body(const uint8_t *payload, + faith_body_size_t payload_size, + faith_envl_cts_event_ack_t *out); diff --git a/faithd/src/commands/conversation.c b/faithd/src/commands/conversation.c index 42722da..f1acf17 100644 --- a/faithd/src/commands/conversation.c +++ b/faithd/src/commands/conversation.c @@ -1,5 +1,25 @@ #include "conversation.h" +#include "../delivery/events.h" + +static faith_status_code_t send_conversation_created( + server_state_t *s, client_conn_t *cl, + const faith_event_conversation_created_t *conv_created) { + + if (!s || !cl || !conv_created) + return FAITH_ERR_INVALID; + uint8_t data[FAITH_EVENT_CONVERSATION_CREATED_DATA_SIZE] = {0}; + faith_body_size_t data_size = 0; + + _FH_CHECK_RETURN(faith_encode_event_conversation_created( + data, &data_size, sizeof(data), conv_created)); + + _FH_CHECK_RETURN(delivery_queue_event(s, cl, FAITH_EVENT_CONVERSATION_CREATED, + data, data_size)); + + return FAITH_OK; +} + faith_status_code_t conv_handle_create_conversation( server_state_t *s, client_conn_t *cl, const faith_envl_cts_command_t *cmd, faith_command_result_t *o_result, faith_command_result_err_t *o_err) { @@ -9,24 +29,30 @@ faith_status_code_t conv_handle_create_conversation( *o_result = FAITH_COMMAND_RESULT_NONE; *o_err = FAITH_COMMAND_ERR_NONE; + faith_status_code_t _fh_result = FAITH_OK; + faith_command_result_err_t potential_err = FAITH_COMMAND_ERR_BAD_COMMAND; + faith_cmd_create_converstation_t create_conv_cmd = {0}; { - _FH_CHECK(faith_decode_cmd_create_conversation( + _FH_CHECK_DEFER(faith_decode_cmd_create_conversation( cmd->payload, cmd->payload_size, &create_conv_cmd)); - if (_fh_rc != FAITH_OK) { - *o_result = FAITH_COMMAND_RESULT_REJECTED; - *o_err = FAITH_COMMAND_ERR_BAD_COMMAND; - return _fh_rc; - } } - char auth_id_hex[33]; - _FH_CHECK( - faith_id128_to_hex(create_conv_cmd.conversant_id.bytes, auth_id_hex)); - printf("Someone wants to chat with: %s\n", auth_id_hex); + faith_conversation_id_t conv_id = {0}; + + potential_err = FAITH_COMMAND_ERR_INTERNAL_ERROR; + _FH_CHECK_DEFER(faith_random_bytes(conv_id.bytes, sizeof(conv_id.bytes))); + + faith_event_conversation_created_t conv_created = {.conversation_id = + conv_id}; + _FH_CHECK_RETURN(send_conversation_created(s, cl, &conv_created)); *o_result = FAITH_COMMAND_RESULT_ACCEPTED; - *o_err = FAITH_COMMAND_ERR_NONE; return FAITH_OK; + +defer: + *o_result = FAITH_COMMAND_RESULT_REJECTED; + *o_err = potential_err; + return _fh_result; } diff --git a/faithd/src/core/core.h b/faithd/src/core/core.h index adca623..971b38f 100644 --- a/faithd/src/core/core.h +++ b/faithd/src/core/core.h @@ -93,7 +93,8 @@ X(FAITH_ERR_ALREADY_CONNECTED, 23) \ X(FAITH_ERR_ALREADY_EXISTS, 24) \ X(FAITH_ERR_ALREADY_REMOVED, 25) \ - X(FAITH_ERR_EPOLL, 26) + X(FAITH_ERR_EPOLL, 26) \ + X(FAITH_ERR_RESYNC_REQUIRED, 27) typedef enum { #define X(name, value) name = value, diff --git a/faithd/src/delivery/event_inbox.c b/faithd/src/delivery/event_inbox.c new file mode 100644 index 0000000..287309b --- /dev/null +++ b/faithd/src/delivery/event_inbox.c @@ -0,0 +1,62 @@ +#include "event_inbox.h" + +#include "../../third_party/stb_ds.h" +#include + +void device_event_inbox_init(device_event_inbox_t* o_inbox) { + if(!o_inbox) return; + + o_inbox->events = NULL; + o_inbox->last_acked_seq = UINT64_MAX; + o_inbox->next_seq = 0; +} + +faith_status_code_t +device_event_inbox_queue_event(device_event_inbox_t *inbox, + const faith_envl_stc_event_t *event) { + if (!inbox || !event) + return FAITH_ERR_INVALID; + arrput(inbox->events, *event); + + return FAITH_OK; +} + +faith_status_code_t device_event_inbox_advance_seq(device_event_inbox_t *inbox) { + if(!inbox) return FAITH_ERR_INVALID; + + /* Dont allow UINT64_MAX */ + if (inbox->next_seq >= UINT64_MAX - 1) + return FAITH_ERR_OVERFLOW; + + inbox->last_sent_seq = inbox->next_seq; + inbox->next_seq++; + + return FAITH_OK; +} + +faith_status_code_t device_event_inbox_ack_seq(device_event_inbox_t *inbox, + uint64_t acked_seq) { + if (!inbox) + return FAITH_ERR_INVALID; + + /* Duplicate or stale ACK */ + if (acked_seq <= inbox->last_acked_seq) + return FAITH_OK; + + if (acked_seq >= inbox->next_seq) + return FAITH_ERR_INVALID; + + inbox->last_acked_seq = acked_seq; + + return FAITH_OK; +} + +faith_status_code_t +device_event_inbox_destroy(device_event_inbox_t *inbox) { + if (!inbox) + return FAITH_ERR_INVALID; + + arrfree(inbox->events); + + return FAITH_OK; +} diff --git a/faithd/src/delivery/event_inbox.h b/faithd/src/delivery/event_inbox.h new file mode 100644 index 0000000..d051d2f --- /dev/null +++ b/faithd/src/delivery/event_inbox.h @@ -0,0 +1,25 @@ +#pragma once + +#include +#include "../codec/events.h" + + +typedef struct { + uint64_t next_seq; + uint64_t last_acked_seq; + uint64_t last_sent_seq; + + faith_envl_stc_event_t* events; +} device_event_inbox_t; + +void device_event_inbox_init(device_event_inbox_t *o_inbox); + +faith_status_code_t +device_event_inbox_queue_event(device_event_inbox_t *inbox, + const faith_envl_stc_event_t *event); + +faith_status_code_t device_event_inbox_advance_seq(device_event_inbox_t *inbox); +faith_status_code_t device_event_inbox_ack_seq(device_event_inbox_t *inbox, + uint64_t acked_seq); + +faith_status_code_t device_event_inbox_destroy(device_event_inbox_t *inbox); diff --git a/faithd/src/delivery/events.c b/faithd/src/delivery/events.c new file mode 100644 index 0000000..afe8c40 --- /dev/null +++ b/faithd/src/delivery/events.c @@ -0,0 +1,75 @@ +#include "events.h" + +#include "../codec/protocol.h" +#include "../server/client_io.h" +#include "event_inbox.h" + +faith_status_code_t delivery_queue_event(server_state_t *s, client_conn_t *cl, + faith_event_codec_type_t type, + uint8_t *data, + faith_body_size_t data_size) { + if (!cl || (!data && data_size != 0) || (data && data_size == 0)) + return FAITH_ERR_INVALID; + + if(!cl->authorized) return FAITH_ERR_UNAUTHORIZED; + + client_device_session_data_t *sess = NULL; + _FH_CHECK_RETURN( + sess_registry_get_session(&s->rt, &cl->auth_id, &cl->device_id, &sess)); + + if(!sess) return FAITH_ERR_UNAUTHORIZED; + + + faith_envl_stc_event_t event = {0}; + event.type = type; + event.data = data; + event.data_size = data_size; + event.seq_num = sess->inbox.next_seq; + + faith_status_code_t _fh_result = FAITH_OK; + + size_t cap = FAITH_ENVL_STC_EVENT_BODY_SIZE_FIXED + data_size; + uint8_t *body = malloc(cap); + faith_body_size_t body_size = 0; + _FH_CHECK_DEFER(faith_encode_event_body(body, &body_size, cap, &event)); + + faith_envelope_t envl = {0}; + envl.type = FAITH_ENVELOPE_EVENT; + envl.recipient_id = cl->auth_id; + envl.body = body; + envl.body_size = body_size; + + _FH_CHECK_DEFER(server_queue_envelope_or_mark_dead(s, cl, &envl)); + + _FH_CHECK_DEFER(device_event_inbox_advance_seq(&sess->inbox)); + +defer: + free(body); + return _fh_result; +} + +faith_status_code_t delivery_handle_event_acked(server_state_t *s, + client_conn_t *cl, + faith_envelope_t *envl) { + if(!s || !cl || !envl) return FAITH_ERR_INVALID; + + if (envl->type != FAITH_ENVELOPE_EVENT_ACK) + return FAITH_ERR_BAD_ENVELOPE; + + if (!cl->authorized) + return FAITH_ERR_UNAUTHORIZED; + + client_device_session_data_t *sess = NULL; + _FH_CHECK_RETURN( + sess_registry_get_session(&s->rt, &cl->auth_id, &cl->device_id, &sess)); + + if(!sess) return FAITH_ERR_UNAUTHORIZED; + + faith_envl_cts_event_ack_t ack = {0}; + _FH_CHECK_RETURN( + faith_decode_event_ack_body(envl->body, envl->body_size, &ack)); + + device_event_inbox_ack_seq(&sess->inbox, ack.seq_num); + + return FAITH_OK; +} diff --git a/faithd/src/delivery/events.h b/faithd/src/delivery/events.h new file mode 100644 index 0000000..04741bd --- /dev/null +++ b/faithd/src/delivery/events.h @@ -0,0 +1,14 @@ +#pragma once + +#include "../codec/events.h" +#include "../core/core.h" +#include "../server/server.h" + +faith_status_code_t delivery_queue_event(server_state_t *s, client_conn_t *cl, + faith_event_codec_type_t type, + uint8_t *data, + faith_body_size_t data_size); + +faith_status_code_t delivery_handle_event_acked(server_state_t *s, + client_conn_t *cl, + faith_envelope_t *envl); diff --git a/faithd/src/server/dispatch.c b/faithd/src/server/dispatch.c index 2c80503..8ecfd78 100644 --- a/faithd/src/server/dispatch.c +++ b/faithd/src/server/dispatch.c @@ -8,37 +8,20 @@ #include "../auth/handshake.h" #include "../delivery/routing.h" +#include "../delivery/events.h" #include "../commands/dispatch.h" -faith_status_code_t server_dispatch_frame(server_state_t *s, client_conn_t *cl, - faith_frame_t *frame) { - if (!s || !cl || !frame) - return FAITH_ERR_INVALID; +static faith_status_code_t server_dispatch_envelope(server_state_t *s, + client_conn_t *cl, + faith_frame_t *frame); - nob_log(INFO, - "[client=%" PRIu64 - " fd=%i] Server got frame: msg_type=%s payload_size=%u", - cl->conn.id, cl->conn.fd, faith_frame_msg_name(frame->msg_type), - frame->payload_size); +static faith_status_code_t +server_handle_ping(server_state_t *s, client_conn_t *cl, faith_frame_t *frame); - switch (frame->msg_type) { - case FAITH_MSG_PING: - _FH_CHECK_RETURN(server_handle_ping(s, cl, frame)); - break; - case FAITH_MSG_ENVL: - _FH_CHECK_RETURN(server_dispatch_envelope(s, cl, frame)); - break; - default: - return FAITH_ERR_BAD_FRAME; - } - - return FAITH_OK; -} - -faith_status_code_t server_dispatch_envelope(server_state_t *s, - client_conn_t *cl, - faith_frame_t *frame) { +static faith_status_code_t server_dispatch_envelope(server_state_t *s, + client_conn_t *cl, + faith_frame_t *frame) { if (!s || !frame || !cl) return FAITH_ERR_INVALID; @@ -75,6 +58,9 @@ faith_status_code_t server_dispatch_envelope(server_state_t *s, case FAITH_ENVELOPE_COMMAND: _FH_CHECK_DEFER(command_dispatch(s, cl, &envl)); break; + case FAITH_ENVELOPE_EVENT_ACK: + _FH_CHECK_RETURN(delivery_handle_event_acked(s, cl, &envl)); + break; default: _FH_RETURN_DEFER(FAITH_ERR_BAD_ENVELOPE); } @@ -92,8 +78,8 @@ faith_status_code_t server_dispatch_envelope(server_state_t *s, return _fh_result; } -faith_status_code_t server_handle_ping(server_state_t *s, client_conn_t *cl, - faith_frame_t *frame) { +static faith_status_code_t +server_handle_ping(server_state_t *s, client_conn_t *cl, faith_frame_t *frame) { if (!s || !cl || cl->closing || !frame) return FAITH_ERR_INVALID; @@ -136,3 +122,28 @@ faith_status_code_t server_handle_ping(server_state_t *s, client_conn_t *cl, return FAITH_OK; } + +faith_status_code_t server_dispatch_frame(server_state_t *s, client_conn_t *cl, + faith_frame_t *frame) { + if (!s || !cl || !frame) + return FAITH_ERR_INVALID; + + nob_log(INFO, + "[client=%" PRIu64 + " fd=%i] Server got frame: msg_type=%s payload_size=%u", + cl->conn.id, cl->conn.fd, faith_frame_msg_name(frame->msg_type), + frame->payload_size); + + switch (frame->msg_type) { + case FAITH_MSG_PING: + _FH_CHECK_RETURN(server_handle_ping(s, cl, frame)); + break; + case FAITH_MSG_ENVL: + _FH_CHECK_RETURN(server_dispatch_envelope(s, cl, frame)); + break; + default: + return FAITH_ERR_BAD_FRAME; + } + + return FAITH_OK; +} diff --git a/faithd/src/server/dispatch.h b/faithd/src/server/dispatch.h index 6d38905..3321e06 100644 --- a/faithd/src/server/dispatch.h +++ b/faithd/src/server/dispatch.h @@ -6,9 +6,3 @@ faith_status_code_t server_dispatch_frame(server_state_t *s, client_conn_t *cl, faith_frame_t *frame); -faith_status_code_t server_dispatch_envelope(server_state_t *s, - client_conn_t *cl, - faith_frame_t *frame); - -faith_status_code_t server_handle_ping(server_state_t *s, client_conn_t *cl, - faith_frame_t *frame); diff --git a/faithd/src/server/sess_registry.c b/faithd/src/server/sess_registry.c index db15d2c..b8f9425 100644 --- a/faithd/src/server/sess_registry.c +++ b/faithd/src/server/sess_registry.c @@ -65,6 +65,9 @@ faith_status_code_t sess_registry_register_session( sess->conn = cl; + /* init event inbox for this device */ + device_event_inbox_init(&sess->inbox); + /* assign public key to new session */ memcpy(sess->ident.public_key, public_key, FAITH_ED25519_PUBLIC_KEY_SIZE); @@ -142,11 +145,11 @@ sess_registry_unregister_session(sess_registry_state_t *rt, free(dev->value); dev->value = NULL; - hmdel(devmap, *device_id); + (void)hmdel(devmap, *device_id); if (hmlen(devmap) == 0) { hmfree(devmap); - hmdel(rt->active_users, *auth_id); + (void)hmdel(rt->active_users, *auth_id); } else { /* write pointer back to avoid stale pointers */ hmput(rt->active_users, *auth_id, devmap); diff --git a/faithd/src/server/sess_registry.h b/faithd/src/server/sess_registry.h index 29b9ec3..dbdac63 100644 --- a/faithd/src/server/sess_registry.h +++ b/faithd/src/server/sess_registry.h @@ -2,6 +2,7 @@ #include "../auth/structs.h" #include "../core/core.h" +#include "../delivery/event_inbox.h" typedef struct { uint8_t public_key[FAITH_ED25519_PUBLIC_KEY_SIZE]; @@ -12,6 +13,7 @@ struct client_conn_t; typedef struct { struct client_conn_t *conn; client_identity_t ident; + device_event_inbox_t inbox; } client_device_session_data_t; typedef struct { diff --git a/faithd/third_party/stb_ds.h b/faithd/third_party/stb_ds.h index e7682b5..abab5b9 100644 --- a/faithd/third_party/stb_ds.h +++ b/faithd/third_party/stb_ds.h @@ -1048,6 +1048,20 @@ typedef int STBDS_SIPHASH_2_4_can_only_be_used_in_64_bit_builds[sizeof(size_t) = #pragma warning(disable:4127) // conditional expression is constant, for do..while(0) and sizeof()== #endif +#if defined(__has_c_attribute) + #if __has_c_attribute(fallthrough) + #define FAITH_FALLTHROUGH [[fallthrough]] + #endif +#endif + +#ifndef FAITH_FALLTHROUGH + #if defined(__GNUC__) && __GNUC__ >= 7 + #define FAITH_FALLTHROUGH __attribute__((fallthrough)) + #else + #define FAITH_FALLTHROUGH ((void)0) + #endif +#endif + static size_t stbds_siphash_bytes(void *p, size_t len, size_t seed) { unsigned char *d = (unsigned char *)p; @@ -1125,27 +1139,27 @@ static size_t stbds_siphash_bytes(void *p, size_t len, size_t seed) #if SIZE_MAX > UINT32_MAX case 7: data |= (size_t)d[6] << 48; - /* fall through */ + FAITH_FALLTHROUGH; case 6: data |= (size_t)d[5] << 40; - /* fall through */ + FAITH_FALLTHROUGH; case 5: data |= (size_t)d[4] << 32; - /* fall through */ + FAITH_FALLTHROUGH; #endif case 4: data |= (size_t)d[3] << 24; - /* fall through */ + FAITH_FALLTHROUGH; case 3: data |= (size_t)d[2] << 16; - /* fall through */ + FAITH_FALLTHROUGH; case 2: data |= (size_t)d[1] << 8; - /* fall through */ + FAITH_FALLTHROUGH; case 1: data |= (size_t)d[0]; - /* fall through */ + FAITH_FALLTHROUGH; case 0: break; }