Keep requests queued by callbacks for the next scheduler batch

src/chat/chat_sync.c

@@ -56,6 +56,7 @@ struct pending_sync {
     struct pending_sync* next;
     uint64_t channel_id, peer_id;
     const char* reason;          /* статическая строка вызывающего обработчика */
+    uint8_t in_batch;            /* задание присутствовало в начале текущего прохода таймера */
 };
 
 struct join_pending {
@@ -490,10 +491,9 @@ static void cs_sync_timer_cb(void* arg) {
     struct chat_sync* cs = arg;
     cs->sync_timer = NULL;
     if (!cs->initialized) return;
-    size_t count = 0;
-    for (struct pending_sync* job = cs->sync_pending; job; job = job->next) count++;
+    for (struct pending_sync* job = cs->sync_pending; job; job = job->next) job->in_batch = 1;
     cs->sync_draining = 1;
-    while (count-- && cs->sync_pending) {
+    while (cs->sync_pending && cs->sync_pending->in_batch) {
         struct pending_sync* job = cs->sync_pending;
         cs->sync_pending = job->next;
         uint64_t peer = job->peer_id, channel_id = job->channel_id;
@@ -618,7 +618,6 @@ static void cs_on_peer_status_changed(struct chat_sync* cs, uint64_t peer, int o
 
     DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: peer_status_changed peer=0x%016llx online=%d channels=%d found=%d",
                 CS_ID, (unsigned long long)peer, online, cs->channel_count, found_in_channel);
-
 }
 
 static void cs_resume_db_sync(struct chat_sync* cs, const char* ch_id) {

tests/test_chat_sync_schedule.c

@@ -11,7 +11,7 @@
 
 static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg);
 static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry);
-static void* schedule_timeout(struct UASYNC* ua, uint64_t tb, void* arg, void (*cb)(void*), const char* name);
+static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name);
 #define member_sync_start observe_start
 #define etcp_send discard_send
 #define uasync_set_timeout schedule_timeout
@@ -51,7 +51,7 @@ static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry) {
     return 0;
 }
 
-static void* schedule_timeout(struct UASYNC* ua, uint64_t tb, void* arg, void (*cb)(void*), const char* name) {
+static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name) {
     if (fail_timer && !strcmp(name, "cs_sync")) { fail_timer = 0; return NULL; }
     return uasync_set_timeout(ua, tb, arg, cb, name);
 }
@@ -193,9 +193,10 @@ int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
     debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG);
     test_clock_init(1000);
+    size_t baseline = u_get_allocated_count();
     test_reinit_isolation(); test_coalesce_and_deadline(); test_sleep_and_cancel();
     test_join_existing_transport(); test_reentrant_cancel(); test_lifecycle_and_timer_failure();
     test_clock_finish();
-    assert(u_get_allocated_count() == u_get_freed_count());
+    assert(u_get_allocated_count() == baseline);
     puts("ALL PASS: targeted chat synchronization scheduling"); return 0;
 }