jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Retry rejected storage with a fresh durable assignment

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:55:34 +01:00
1 parent 9edc06cb09
commit bfc8e26431
2 files changed
+30 -15
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox_media.c
+30 -13
View File
@@ -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);
src/dm/dm_media.c
-2
View File
@@ -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;