Retain DM coordinator bodies when assignment scans fail

This commit is contained in:
evgeny committed 2026-10-26 19:42:00 +01:00
1 parent 2fce4641ec
commit 3f195c058b
3 files changed
+28 -9

No files matched your search

+26 -9
View File
@@ -182,10 +182,17 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
(unsigned long long)source, (unsigned long long)expiry);
} else if ((cmd == DM_MEDIA_DELETE_ACK || cmd == DM_MEDIA_EXPIRED) && len == 16) {
if (cmd == DM_MEDIA_EXPIRED && !delivered) { dm_job_release(media, p, source, 1); return; }
st = dm_media_prepare(inst, "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=?");
st = dm_media_prepare(inst, "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=? AND released=0");
if (!st) return;
sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)source);
dm_media_step(inst, st);
rc = sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)source);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: ACK bind failed M=%llu cmd=%u rc=%d",
(unsigned long long)source, cmd, rc);
sqlite3_finalize(st); return;
}
if (!dm_media_step(inst, st) && sqlite3_changes(inst->topo_sqlite_db))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: assignment released M=%llu cmd=%u", (unsigned long long)source, cmd);
} else if (cmd == DM_MEDIA_REJECT && len == 17 && !delivered) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: storage refused M=%llu reason=%u", (unsigned long long)source, p[16]);
dm_job_release(media, p, source, 0);
@@ -220,20 +227,30 @@ static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], cons
struct UTUN_INSTANCE* inst = media->inst;
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT node_id,group_id FROM dm_media_holders WHERE id=? AND released=0 LIMIT 16");
if (!st) return;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup scan bind failed rc=%d; body retained", rc);
sqlite3_finalize(st); return;
}
uint64_t peers[16], groups[16];
unsigned count = 0;
int rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW && count < 16) {
peers[count] = (uint64_t)sqlite3_column_int64(st, 0); groups[count++] = (uint64_t)sqlite3_column_int64(st, 1);
}
sqlite3_finalize(st);
if (rc != SQLITE_ROW && rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup scan failed rc=%d", rc); return; }
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup scan failed rc=%d finalize=%d; body retained", rc, final); return;
}
for (unsigned i = 0; i < count; i++) dm_media_send(inst, groups[i], peers[i], DM_MEDIA_DELETE, receipt, DM_MEDIA_RECEIPT_SIZE);
if (!count) {
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET body=NULL WHERE id=? AND receipt IS NOT NULL AND body IS NOT NULL");
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET body=NULL WHERE id=? AND receipt IS NOT NULL AND body IS NOT NULL"
" AND NOT EXISTS(SELECT 1 FROM dm_media_holders h WHERE h.id=dm_media_jobs.id AND h.released=0)");
if (!st) return;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup body bind failed rc=%d; retry retained", rc);
sqlite3_finalize(st); return;
}
if (!dm_media_step(inst, st) && sqlite3_changes(inst->topo_sqlite_db))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: all storage assignments released; message body removed");
}
+1
View File
@@ -26,6 +26,7 @@ int dm_mailbox_media_register(struct dm_mb_media* media, uint64_t group, uint64_
* STORED/REJECT/ACK относятся к сохранённому назначению M; RECEIPT проверяется отдельно подписью B.
* EXPIRED/REJECT атомарно освобождают назначение и сбрасывают текущий holder; ошибка откатывает оба изменения.
* Таймер завершает stored-назначение по expires даже при offline B/потерянном EXPIRED, также после restart.
* После квитанции тело удаляется только при отсутствии неосвобождённых назначений; ошибка сканирования откладывает очистку.
* Для неподтверждённого STORE повтор запрашивает результат из custody/tombstone M; квитанция B не отменяется. */
void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t source,
uint8_t cmd, const uint8_t* body, size_t len);
+1
View File
@@ -8,6 +8,7 @@
#include "../src/dm/dm_media.h"
#include "test_clock.h"
#include <assert.h>
#include <stdlib.h>
#include <string.h>
enum operation { NO_FAULT, PREPARE, BIND_BLOB, BIND_INT, STEP, FINALIZE };