diff --git a/src/dm/dm_mailbox_media.c b/src/dm/dm_mailbox_media.c index 234ee781..22a413f2 100644 --- a/src/dm/dm_mailbox_media.c +++ b/src/dm/dm_mailbox_media.c @@ -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; } diff --git a/src/dm/dm_mailbox_media.h b/src/dm/dm_mailbox_media.h index bd0f5e32..464224e4 100644 --- a/src/dm/dm_mailbox_media.h +++ b/src/dm/dm_mailbox_media.h @@ -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 общей группы.