From bfc8e264314a5eb57038cf60cd7762071e5dbeea Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 22:55:34 +0100 Subject: [PATCH] Retry rejected storage with a fresh durable assignment --- src/dm/dm_mailbox_media.c | 43 +++++++++++++++++++++++++++------------ src/dm/dm_media.c | 2 -- 2 files changed, 30 insertions(+), 15 deletions(-) diff --git a/src/dm/dm_mailbox_media.c b/src/dm/dm_mailbox_media.c index 5761fde8c..4e2e5408b 100644 --- a/src/dm/dm_mailbox_media.c +++ b/src/dm/dm_mailbox_media.c @@ -121,7 +121,7 @@ static int dm_job_release(struct dm_mb_media* media, const uint8_t id[16], uint6 "UPDATE dm_media_holders SET released=3 WHERE id=? AND node_id=? AND assignment_id=? AND released=0", expired ? "UPDATE dm_media_jobs SET holder=0,stored=0,expired=1,retry_at=0 WHERE id=?1 AND holder=?2 AND receipt IS NULL" " AND EXISTS(SELECT 1 FROM dm_media_holders h WHERE h.id=?1 AND h.node_id=?2 AND h.assignment_id=?3)" : - "UPDATE dm_media_jobs SET holder=0,stored=0,expires=0,delete_on_open=-1,retry_at=0 WHERE id=?1 AND holder=?2 AND receipt IS NULL" + "UPDATE dm_media_jobs SET holder=0,stored=0,expires=0,delete_on_open=-1,retry_at=?4 WHERE id=?1 AND holder=?2 AND receipt IS NULL" " AND EXISTS(SELECT 1 FROM dm_media_holders h WHERE h.id=?1 AND h.node_id=?2 AND h.assignment_id=?3)" }; int changed[2]; for (int i = 0; i < 2; i++) { @@ -130,6 +130,7 @@ static int dm_job_release(struct dm_mb_media* media, const uint8_t id[16], uint6 int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)holder); if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC); + if (rc == SQLITE_OK && i == 1 && !expired) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)(ntp_time_get_seconds(inst) + 60)); if (rc == SQLITE_OK) rc = sqlite3_step(st); int final = sqlite3_finalize(st); if (rc != SQLITE_DONE || final != SQLITE_OK) { @@ -223,7 +224,7 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s } else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: unexpected control cmd=%u bytes=%zu", cmd, len); } -/* Выбрать один достижимый storage, исключив прежние отказы/истёкшие назначения. */ +/* Сначала новый M, затем новая попытка после отказа по квоте. Истёкшие копии не возрождаем. */ static uint64_t dm_job_storage(struct dm_mb_media* media, const uint8_t id[16]) { struct UTUN_INSTANCE* inst = media->inst; uint64_t* groups = NULL; @@ -232,15 +233,17 @@ static uint64_t dm_job_storage(struct dm_mb_media* media, const uint8_t id[16]) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM storage: membership lookup failed"); return 0; } uint64_t found = 0; - for (int i = 0; i < count && !found; i++) { + for (int retry = 0; retry < 2 && !found; retry++) for (int i = 0; i < count && !found; i++) { if (!topo_group_chat_allowed(inst, groups[i], inst->node_id)) continue; - char sql[320]; - snprintf(sql, sizeof(sql), "SELECT node_id,adm_tags FROM \"peers_%llu\"" - " WHERE source=0 AND deleted=0 AND admitted=1 AND node_id NOT IN" - " (SELECT node_id FROM dm_media_holders WHERE id=?) ORDER BY node_id", (unsigned long long)groups[i]); + char sql[384]; + snprintf(sql, sizeof(sql), "SELECT p.node_id,p.adm_tags FROM \"peers_%llu\" p" + " LEFT JOIN dm_media_holders h ON h.node_id=p.node_id AND h.id=?1" + " WHERE p.source=0 AND p.deleted=0 AND p.admitted=1 AND COALESCE(h.released,-1)=?2 ORDER BY p.node_id", + (unsigned long long)groups[i]); sqlite3_stmt* st = dm_media_prepare(inst, sql); - if (!st) break; + if (!st) goto done; int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 2, retry ? 3 : -1); while (rc == SQLITE_OK && (rc = sqlite3_step(st)) == SQLITE_ROW) { uint64_t node = (uint64_t)sqlite3_column_int64(st, 0), route = 0; const char* tags = (const char*)sqlite3_column_text(st, 1); @@ -251,9 +254,10 @@ static uint64_t dm_job_storage(struct dm_mb_media* media, const uint8_t id[16]) } int final = sqlite3_finalize(st); if ((rc != SQLITE_OK && rc != SQLITE_DONE) || final != SQLITE_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM storage: candidate scan failed rc=%d final=%d", rc, final); found = 0; break; + DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM storage: candidate scan failed rc=%d final=%d", rc, final); found = 0; goto done; } } +done: u_free(groups); return found; } @@ -430,16 +434,29 @@ static void dm_mailbox_media_tick(void* arg) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: random ID generation failed"); continue; } if (dm_media_exec(inst, "BEGIN IMMEDIATE")) continue; - st = dm_media_prepare(inst, "INSERT INTO dm_media_holders(id,node_id,assignment_id) VALUES(?,?,?)"); + st = dm_media_prepare(inst, "INSERT INTO dm_media_holders(id,node_id,assignment_id) VALUES(?,?,?)" + " ON CONFLICT(id,node_id) DO UPDATE SET assignment_id=excluded.assignment_id,released=0 WHERE released=3"); if (!st) { dm_media_exec(inst, "ROLLBACK"); continue; } - sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder); - sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC); + rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)holder); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC); + if (rc != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: bind failed rc=%d", rc); + sqlite3_finalize(st); dm_media_exec(inst, "ROLLBACK"); continue; + } if (dm_media_step(inst, st)) { dm_media_exec(inst, "ROLLBACK"); continue; } + if (sqlite3_changes(inst->topo_sqlite_db) != 1) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: cannot replace unresolved assignment M=%llu", (unsigned long long)holder); + dm_media_exec(inst, "ROLLBACK"); continue; + } st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET holder=?,stored=0 WHERE id=? AND receipt IS NULL"); if (!st) { dm_media_exec(inst, "ROLLBACK"); continue; } sqlite3_bind_int64(st, 1, (sqlite3_int64)holder); sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC); if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) { dm_media_exec(inst, "ROLLBACK"); continue; } - DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: assigned storage M=%llu", (unsigned long long)holder); + uint64_t token[2], file[2]; memcpy(token, assignment, 16); memcpy(file, id, 16); + DEBUG_INFO(DEBUG_CATEGORY_DM, "DM assignment: committed assignment=%016llx%016llx file=%016llx%016llx M=%llu", + (unsigned long long)token[0], (unsigned long long)token[1], (unsigned long long)file[0], + (unsigned long long)file[1], (unsigned long long)holder); } uint8_t put[24 + DM_MEDIA_BODY_MAX]; memcpy(put, assignment, 16); memcpy(put + 16, &recipient, 8); memcpy(put + 24, body, len); dm_media_send(inst, 0, holder, DM_MEDIA_STORE, put, len + 24); diff --git a/src/dm/dm_media.c b/src/dm/dm_media.c index 34ab30788..d8170da76 100644 --- a/src/dm/dm_media.c +++ b/src/dm/dm_media.c @@ -1082,8 +1082,6 @@ void dm_media_action_trampoline(void* arg) { static void dm_custody_store(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, const uint8_t* p, size_t len) { if (len < 24) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: short STORE"); return; } const uint8_t* assignment = p; - static const uint8_t zero[16]; - if (!memcmp(assignment, zero, 16)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: zero assignment ID"); return; } p += 16; len -= 16; uint64_t recipient, author, size; memcpy(&recipient, p, 8); p += 8; len -= 8;