Expose per-model work policy and checkpoint results for history

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);
+    }
+}

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.