merkle_sync: fix _handle_batch — call _session_done after terminal buckets

tools/chatgui/transport/merkle_sync.c

@@ -487,6 +487,8 @@ static void _handle_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
 
     DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: handle_batch peer=%016llx ns=%s count=%d len=%zu", MS_ID, (unsigned long long)peer, ns, count, plen - 1);
 
+    int all_terminal = 1;
+
     for (uint8_t i = 0; i < count && rem >= 3; i++) {
         uint8_t lvl = bp[0]; uint8_t pb_i = bp[1]; rem -= 2; bp += 2;
         if (rem < pb_i + 1) break;
@@ -494,11 +496,12 @@ static void _handle_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
         uint8_t is_data = *bp++; rem--;
 
         if (is_data && rem >= 2) {
-            uint16_t mc; memcpy(&mc, bp, 2);
-            size_t item_len = rem; /* pass entire remainder to apply_items */
-            ms->ops->apply_items(ms->data_ctx, ns, bp, item_len);
+            uint16_t item_count; memcpy(&item_count, bp, 2);
+            if (rem < 2 + (size_t)item_count * 2) break;
+            ms->ops->apply_items(ms->data_ctx, ns, bp, rem);
             bp += rem; rem = 0;
         } else if (!is_data && rem >= 4) {
+            all_terminal = 0;
             uint8_t sub_pl[4096]; size_t sub_len = 0;
             uint8_t next_lvl = (uint8_t)(lvl < MT_MAX_LEVEL ? lvl + 1 : lvl);
             sub_pl[sub_len++] = next_lvl;
@@ -516,6 +519,10 @@ static void _handle_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
             _handle_hashes(ms, peer, ns, sub_pl, sub_len);
         }
     }
+    if (all_terminal) {
+        struct ms_session* s = _session_find(ms, peer, ns);
+        if (s && s->active) _session_done(s, MT_OK);
+    }
 }
 
 static void _recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {