Reconcile DM storage expiry and release assignments atomically

This commit is contained in:
evgeny committed 2026-10-26 17:49:26 +01:00
1 parent dda4a780e3
commit cb08faae04
2 files changed
+42 -19

No files matched your search

+39 -16
View File
@@ -110,6 +110,35 @@ static void dm_job_receipt(struct dm_mb_media* media, uint64_t group, uint64_t s
}
}
/* Освобождение назначения и сброс текущего holder — один durable переход. */
static int dm_job_release(struct dm_mb_media* media, const uint8_t id[16], uint64_t holder, int expired) {
struct UTUN_INSTANCE* inst = media->inst;
if (dm_media_exec(inst, "BEGIN IMMEDIATE")) return -1;
sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=? AND released=0");
if (!st) goto rollback;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
if (dm_media_step(inst, st)) goto rollback;
int released = sqlite3_changes(inst->topo_sqlite_db);
const char* sql = expired ?
"UPDATE dm_media_jobs SET holder=0,stored=0,expired=1,retry_at=0 WHERE id=? AND holder=? AND receipt IS NULL" :
"UPDATE dm_media_jobs SET holder=0,stored=0,retry_at=0 WHERE id=? AND holder=? AND receipt IS NULL";
st = dm_media_prepare(inst, sql);
if (!st) goto rollback;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
if (dm_media_step(inst, st)) goto rollback;
int reset = sqlite3_changes(inst->topo_sqlite_db);
if (dm_media_exec(inst, "COMMIT")) goto rollback;
if (released || reset) DEBUG_INFO(DEBUG_CATEGORY_DM,
"dm_mailbox_media: storage released M=%llu reason=%s released=%d reset=%d",
(unsigned long long)holder, expired ? "expired" : "rejected", released, reset);
return 0;
rollback:
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: storage release rolled back M=%llu reason=%s; retry retained",
(unsigned long long)holder, expired ? "expired" : "rejected");
dm_media_exec(inst, "ROLLBACK");
return -1;
}
/* Control от M принимается только для durable зарегистрированного назначения. */
void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t source,
uint8_t cmd, const uint8_t* p, size_t len) {
@@ -147,26 +176,14 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: custody committed M=%llu expires=%llu",
(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=?");
if (!st) return;
sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)source);
if (dm_media_step(inst, st)) return;
if (!delivered && cmd == DM_MEDIA_EXPIRED) {
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET holder=0,stored=0,expired=1,retry_at=0 WHERE id=? AND holder=?");
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);
}
} 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]);
st = dm_media_prepare(inst, "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)source);
if (dm_media_step(inst, st)) return;
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET holder=0,stored=0,retry_at=0 WHERE id=? AND holder=?");
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);
} 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);
} else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: unexpected control cmd=%u bytes=%zu", cmd, len);
}
@@ -254,6 +271,12 @@ static void dm_mailbox_media_tick(void* arg) {
if (dm_route_group(inst, author, group, &gid)) dm_media_send(inst, gid, author, DM_MEDIA_RECEIPT, receipt, sizeof(receipt));
continue;
}
if (holder && stored && expires <= now) {
if (!dm_job_release(media, id, holder, 1))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: custody deadline reconciled M=%llu expires=%llu now=%llu",
(unsigned long long)holder, (unsigned long long)expires, (unsigned long long)now);
continue; /* Новый снимок после commit; при ошибке задача повторится через retry_at. */
}
if (!len || len > sizeof(body) || !dm_message_media(body, len)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: corrupt saved descriptor"); continue;
}
+3 -3
View File
@@ -24,9 +24,9 @@ int dm_mailbox_media_register(struct dm_mb_media* media, uint64_t group, uint64_
/* После проверки router-подписи и размера команды в dm_media: group/source — доверенные поля router.
* body заимствован на время вызова; RECEIPT=152, STORED=24, DELETE_ACK/EXPIRED=16, REJECT=17 байт.
* STORED/REJECT/ACK относятся к сохранённому назначению M; RECEIPT проверяется отдельно подписью B.
* SQL-ошибки логируются; отдельные обновления здесь не объединены в транзакцию.
* Известный дефект: отказ сброса job после EXPIRED оставляет stored/holder при released=1;
* пока B offline, таймер не восстанавливает переход. См. doc/dm_audit_findings.txt, раздел10. */
* 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);
/* Проверить 152-байтовую квитанцию и попытаться отправить доступным S общей группы.