diff --git a/src/chat/merkle_sync.c b/src/chat/merkle_sync.c index bcf701c0..dfcb7dce 100644 --- a/src/chat/merkle_sync.c +++ b/src/chat/merkle_sync.c @@ -115,6 +115,9 @@ static void ms_notify(struct ms_session* s, int result) { s->requests = NULL; uint64_t peer = s->peer; char ns[sizeof(s->ns)]; memcpy(ns, s->ns, sizeof(ns)); + void (*on_result)(void*, const char*, uint64_t, int) = s->dataset->ops->on_result; + void* ctx = s->dataset->data_ctx; + if (on_result) on_result(ctx, ns, peer, result); while (r) { struct ms_request* next = r->next; merkle_sync_done_cb cb = r->cb; void* arg = r->arg; @@ -204,6 +207,10 @@ static int ms_root(struct ms_session* s, uint8_t root[MT_HASH_SIZE]) { return merkle_sync_read_hash(s->ms->inst, s->dataset->id, s->ns, 0, 0, root); } +static int ms_can_run(struct merkle_sync* ms, struct ms_dataset* model, struct ETCP_CONN* conn, const char* ns) { + return model->ops->can_run ? model->ops->can_run(model->data_ctx, ns, conn->peer_node_id) == 1 : !conn->peer_sleep_phase; +} + static void ms_sent(struct ll_queue* q, void* arg) { struct ms_session* s = arg; struct ll_entry* entry = s->tx; @@ -211,7 +218,8 @@ static void ms_sent(struct ll_queue* q, void* arg) { if (!entry) return; int type = entry->dgram[9]; size_t len = entry->len; - if (!s->conn->links_up || s->conn->state == 2 || s->conn->close_requested || q != s->conn->send_input_q) { + if (!s->conn->links_up || s->conn->state == 2 || s->conn->close_requested || q != s->conn->send_input_q || + !ms_can_run(s->ms, s->dataset, s->conn, s->ns)) { queue_dgram_free(entry); queue_entry_free(entry); ms_fail(s, MT_ERR_DISCONNECTED, "connection unavailable", 0); return; @@ -363,6 +371,9 @@ static void ms_work(void* arg) { struct ms_session* s = arg; s->work = NULL; if (s->state != MS_IDLE || !s->dirty || s->tx) return; + if (!ms_can_run(s->ms, s->dataset, s->conn, s->ns)) { + ms_fail(s, MT_ERR_DISCONNECTED, "model work deferred", 0); return; + } int rc; if (s->leader) rc = ms_begin(s); else { s->dirty = 0; rc = ms_send(s, MS_WAKE, 0, NULL, 0); } @@ -520,6 +531,11 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: unauthorized ns=%s peer=%016llx", ns, (unsigned long long)conn->peer_node_id); queue_dgram_free(entry); queue_entry_free(entry); return; } + if (!ms_can_run(ms, model, conn, ns)) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: model receive deferred dataset=%u ns=%s peer=%016llx", + dataset, ns, (unsigned long long)conn->peer_node_id); + queue_dgram_free(entry); queue_entry_free(entry); return; + } struct ms_session* s = ms_find(ms, conn, dataset, ns_id); if (!s && (type == MS_WAKE || type == MS_BEGIN)) s = ms_create(ms, conn, dataset, ns_id); if (!s) { @@ -699,7 +715,7 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer uint64_t ns_id; struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL; if (!ms || !ms->datasets[dataset].ops || ms_namespace(ns, &ns_id) < 0 || !conn || !conn->initialized || !conn->links_up || conn->state == 2 - || conn->peer_sleep_phase || peer == inst->node_id || conn == ms->closing_conn) { + || peer == inst->node_id || conn == ms->closing_conn || !ms_can_run(ms, &ms->datasets[dataset], conn, ns)) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: start unavailable ns=%s peer=%016llx", ns ? ns : "-", (unsigned long long)peer); return -1; } @@ -745,3 +761,37 @@ void merkle_sync_cancel_peer(struct UTUN_INSTANCE* inst, uint64_t peer) { struct ms_session* s = ms->sessions; while (s) { struct ms_session* next = s->next; if (s->peer == peer) ms_free(s); s = next; } } + +void merkle_sync_cancel_dataset_ns(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns) { + struct merkle_sync* ms = inst ? inst->msync : NULL; + if (!ms || !ns) return; + struct ms_session* s = ms->sessions; + while (s) { + struct ms_session* next = s->next; + if (s->dataset->id == dataset && !strcmp(s->ns, ns)) ms_free(s); + s = next; + } +} + +void merkle_sync_cancel_dataset_peer(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer) { + struct merkle_sync* ms = inst ? inst->msync : NULL; + if (!ms) return; + struct ms_session* s = ms->sessions; + while (s) { + struct ms_session* next = s->next; + if (s->dataset->id == dataset && s->peer == peer) ms_free(s); + s = next; + } +} + +void merkle_sync_recheck_peer(struct UTUN_INSTANCE* inst, uint64_t peer) { + for (;;) { + struct merkle_sync* ms = inst ? inst->msync : NULL; + if (!ms) return; + struct ms_session* selected = NULL; + for (struct ms_session* s = ms->sessions; s; s = s->next) + if (s->peer == peer && s->state != MS_FAILED && !ms_can_run(ms, s->dataset, s->conn, s->ns)) { selected = s; break; } + if (!selected) return; + ms_fail(selected, MT_ERR_DISCONNECTED, "model policy changed", 0); + } +} diff --git a/src/chat/merkle_sync.h b/src/chat/merkle_sync.h index 4d3d1959..5591241d 100644 --- a/src/chat/merkle_sync.h +++ b/src/chat/merkle_sync.h @@ -24,6 +24,7 @@ struct UTUN_INSTANCE; #define MT_KEY_SIZE 32 #define MT_DATA_MEMBERS 0 #define MT_DATA_FEEDBACK 1 +#define MT_DATA_MESSAGES 2 /* Полный ключ курсора, big-endian; первые 8 байт задают ветвь дерева. * Для node_id участника: BE64(node_id), затем 24 нулевых байта. */ @@ -43,6 +44,11 @@ struct merkle_sync_data_ops { * <0 — ошибка. Никаких broadcast/send-back из этого callback. */ int (*apply_items)(void* ctx, const char* ns, uint64_t from_peer, const uint8_t* data, size_t len); int (*validate_peer)(void* ctx, const char* ns, uint64_t peer); // необязательная проверка: только 1 разрешает приём + /* Политика работы (например sleep/sound), 1 разрешает обмен; без callback спящий пир отложен. */ + int (*can_run)(void* ctx, const char* ns, uint64_t peer); + /* Результат каждой контрольной точки/ошибки, включая автоматические раунды. + * Вызывается до пользовательских done callbacks; может отменить сессию. */ + void (*on_result)(void* ctx, const char* ns, uint64_t peer, int result); /* Проверить отложенные зависимости модели после обоих проходов, до подтверждения корня. */ int (*finish)(void* ctx, const char* ns); /* Необязательный отложенный приём: все четыре callback задаются вместе. @@ -90,6 +96,11 @@ void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t pe /* Отменить все сессии namespace / пира без callbacks. */ void merkle_sync_cancel_ns(struct UTUN_INSTANCE* inst, const char* ns); void merkle_sync_cancel_peer(struct UTUN_INSTANCE* inst, uint64_t peer); +/* Отменить только указанный dataset; остальные модели namespace/пира продолжают работу. */ +void merkle_sync_cancel_dataset_ns(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns); +void merkle_sync_cancel_dataset_peer(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer); +/* После изменения sleep/sound завершить запрещённые попытки, сохраняя подписки для повтора. */ +void merkle_sync_recheck_peer(struct UTUN_INSTANCE* inst, uint64_t peer); /* Обновление индекса вызывается моделью данных. * При внешней транзакции caller вызывает changed ПОСЛЕ commit.