diff --git a/doc/etcp_multipath_plan.txt b/doc/etcp_multipath_plan.txt index 00bc89f0..b16195b5 100644 --- a/doc/etcp_multipath_plan.txt +++ b/doc/etcp_multipath_plan.txt @@ -3,6 +3,14 @@ ETCP: независимая обратная связь каналов и до Статус +24.10.2026, объединение с актуальным master: удаление чужого member_files temp +уже исправлено0f42f92f, ранняя отмена —20b4975d; повторный фикс не требуется. +Описание ниже185/186 относится к прежней базе a198cfb9, до этих исправлений. +Актуальный аудит и проверки: doc/member_files_audit_findings.txt. +Небольшой недобор до условных85% — диагностическая метрика; он сам по себе +не оправдывает усложнения BBR. Дальше приоритет STCP backpressure, смена RTT/BW +нескольких путей и повторный отказ с проверкой доставки, пауз и роста очередей. + 24.10.2026, продолжение после push: отдельно исследован одиночный дуплекс2Мбит/с. Все прежние15 локальных коммитов вошли в origin/master после rebase; патчи сохранены. Из-за параллельных изменений основного каталога продолжение изолировано в ветке diff --git a/doc/member_files.txt b/doc/member_files.txt index f61ba5bf..12aa3c9f 100644 --- a/doc/member_files.txt +++ b/doc/member_files.txt @@ -25,7 +25,8 @@ member_files_path=/путь/к/хранилищу (опционально, до 255 байт UTF-8) Путь, квота и TTL читаются при старте чат-сервиса; изменение требует его restart. -enabled проверяется на каждом запросе. Лимит одинаков для каждого UID отдельно, +enabled проверяется на каждом серверном запросе; клиент доступа к другим storage остаётся доступен. +Лимит одинаков для каждого UID отдельно, размер UID суммируется по всем группам на этом сервере. Значение в МиБ (1024*1024 байт). Счётчики used/reserved/files/directories хранятся в памяти дискового worker и восстанавливаются с диска только при старте. Каталоги не занимают байтовую квоту. files/directories считают опубликованные файлы и каталоги внутри деревьев UID @@ -86,7 +87,7 @@ Wire, версия 4 Данные максимум 8192 байт; вместе с полным router-путём и crypto укладываются в лимит normalizer 16384 байта. Размер файла/смещение ограничены INT64_MAX. Пути — строгий UTF-8, относительные, разделитель '/'. Запрещены пустые компоненты, -компоненты с ведущей '.', завершающей точкой/пробелом, управляющие символы, +компоненты с ведущей '.', завершающей точкой/пробелом, ASCII C0/DEL, Windows device names и символы \ : * ? " < > |. Корень внутри UID — пустой путь. POSIX разрешает путь через *at/O_NOFOLLOW; regular-файлы с hardlink отвергаются. Windows удерживает родительские каталоги без SHARE_DELETE и не принимает reparse points. @@ -168,11 +169,19 @@ Windows удерживает родительские каталоги без SH Статусы (0..14): OK, ACCESS_DENIED, NOT_FOUND, INVALID_PATH, INVALID_RANGE, EXISTS, BUSY, STALE, NO_SPACE, IO_ERROR, BAD_PACKET, HASH_MISMATCH, CANCELLED, TIMEOUT, DIRECTORY_NOT_EMPTY. -Мутации COMMIT/MKDIR/RENAME/REMOVE сохраняют успешный ответ на 120 секунд +Мутации COMMIT/MKDIR/RENAME/REMOVE сохраняют успешный ответ до истечения 120 секунд от начала обработки (64 квитанции); тот же request_id с другим содержимым — BAD_PACKET. COMMIT дополнительно узнаётся по upload UUID. При заполнении квитанций новая -мутация возвращает BUSY до истечения старых. Клиент повторяет запрос с прежним -request_id раз в 5 секунд, общий срок 120 секунд. Повторы после этого срока +мутация возвращает BUSY до истечения старых. В таком ответе cursor содержит +остаток до ближайшего освобождения квитанции в timebase 0.1мс (1..1200000). +Нулевой cursor означает BUSY другой причины с обычной задержкой 5 секунд. +Клиент COMMIT/MKDIR/RENAME/REMOVE сохраняет pending и прежний request_id/параметры, +ставит отменяемый таймер на этот срок и повторяет команду; промежуточный BUSY не завершает callback. +При потере ответа обычный повтор идёт через 5 секунд. Общий срок от создания +запроса остаётся 120 секунд и не продлевается даже при BUSY; pump не отправляет +команду после этого срока. Поэтому повторы остаются внутри серверного окна +квитанции, не требуя её досрочного вытеснения. Остальные команды возвращают BUSY caller. +Повторы после этого срока не дают гарантии того же результата; перед новой мутацией требуется STAT. Потоки, таймауты и остановка @@ -224,6 +233,8 @@ quiesce запрещает новые команды, отменяет тайм вне callbacks дожидается только своих работ/completions и закрывает дисковый поток до уничтожения CHAT-групп. Контексты не освобождаются под работающим I/O. Отмена не прерывает текущий системный I/O или хеширование, а ждёт его completion. +Немедленная отмена после start снимает starting: поздний CM UP не запускает STAT/подготовку заново. +После перехода к финализации публикации повторный cancel не меняет уже выбранный результат. Desktop ------- @@ -262,7 +273,14 @@ Stop сервиса/группы завершает все принадлежа Все API/callbacks выполняются в uasync-потоке экземпляра. GUI trampolines используют тот же API и превращают callback в существующие события прогресса; Qt не делает I/O. -Один локальный файл назначения не может одновременно принадлежать двум download. +Два download с одной строкой local одновременно не допускаются; физические aliases +и подмену локальных путей исключает вызывающий. Upload читает стабильный regular source; +изменение снимка возвращает STALE. На POSIX O_NONBLOCK позволяет отвергнуть FIFO +до блокирующего open/read единственного дискового worker. +Download создаёт соседний temp эксклюзивно и получает право cleanup только после успеха open. +Отказ из-за существующего temp, max_bytes, STALE или ранняя отмена не удаляют чужой путь. +После проверки SHA256 выполняются sync/close и заменяющий rename; до публикации прежний local сохраняется. +POSIX-публикация локального download не делает fsync родительского каталога, в отличие от серверного COMMIT. Аватарки пользователей --------------------- @@ -350,12 +368,18 @@ tests/test_member_files: wire/UTF-8, произвольные блоки, пов резервирование/повторы/TTL, инкрементальные files/directories, замены пустых файлов, создание/удаление/rename каталогов, отсутствие runtime-rescan, стартовый scan и его ошибки, ссылки, expiry/restart, 10-секундный fd idle и принадлежность work/done потокам. +На POSIX отдельный процесс с границей 2с проверяет неблокирующее открытие FIFO +и последующий отказ проверки типа; fork выполняется до создания pthread. tests/test_member_files_network: настоящие A/B/M, подписанная storage-роль, параллельные upload/download одного пира, local loopback, пустые файлы, побайтная проверка, отмена одного handle, restart router и destroy с активным server I/O. Шестисекундная задержка worker проверяет receive backpressure и busy ACK. Также проверяются чужая запись, виртуальный корень и подписанное удаление читателя. QUOTA проверяется на настоящем подписанном/зашифрованном канале; удалённый читатель не получает счётчики. +Проверены сохранность чужого temp/прежнего назначения при EXISTS, max_bytes и немедленном cancel, +поздний CM UP после отмены, 76 созданий и 76 удалений с исчерпанием квитанций. +Один успешный ответ отброшен перед клиентским handler: повтор с прежним ID возвращает квитанцию, +счётчики остаются точными (76 каталогов после создания, 0 после удаления). test_services — stop/start общего чат-сервиса. Диагностика файлового сервиса — категория media, маршрутизатора — etcp_route. Настройка для полного трасса этих модулей: diff --git a/doc/member_files_audit_findings.txt b/doc/member_files_audit_findings.txt new file mode 100644 index 00000000..51c5e989 --- /dev/null +++ b/doc/member_files_audit_findings.txt @@ -0,0 +1,128 @@ +Аудит C-модуля member_files — 2026-10-24 + +Статус: полностью прочитаны шесть реализаций и шесть заголовков src/member_files/. +Заголовки сверены; четыре подтверждённых дефекта исправлены и проверены. +Серверная/клиентская части doc/member_files.txt согласованы с кодом. Общий документ +остаётся [~]: его desktop/AvatarManager/Android-подсистемы целиком в этот аудит не входили. +Ниже результаты относятся к C-сервису, а не к полному аудиту всех файлов GUI/аватарок. + +1. Удаление чужого локального temp + +До исправления TRANSFER_FINISH безусловно удалял вычисленный путь temp при ошибке, +включая отказ эксклюзивного open из-за существующего файла и отказ до открытия. +Настоящий test_member_files_network: preexisting target/temp по37 проверенных байт, +скачивание с подписанного storage, MF_EXISTS; target=1, foreign_temp=0, FAIL phase23. +Корректное воспроизведение: /tmp/member-files-temp-repro5.log, строки391..395. + +Исправление: один temp_owned у handle, устанавливается только после успешного +эксклюзивного создания. Finish закрывает fd, публикует/удаляет только свой temp. +После успешного rename владение именем temp снято; source/target не подменяются caller. +Новый FILE-слой или второй файловый владелец не добавлены: disk API работает с native fd. +Обычный и sanitizer-прогоны: EXISTS, max_bytes, немедленная отмена сохраняют +прежние37 байт target и чужого temp, finished ровно один раз. + +2. Ранняя отмена возобновляла передачу и приводила к UAF + +После cancel сразу вслед за start поздний CM UP вызывал wake. Starting оставался1, +поэтому pump после начала cleanup снова посылал STAT. Finish освобождал handle, +а ответ STAT вызывал transfer_response с уже освобождённым arg. +Лог: /tmp/member-files-temp-fixed.log; GDB: /tmp/member-files-cancel-gdb.log, +SIGSEGV в transfer_response → mf_pending_finish → mf_dispatch. + +Переход stopping теперь снимает starting; wake без готового финального ответа +ничего не планирует. Обязательная cleanup-задача остаётся единственным продолжением, +затем отправляется финальный CANCEL/END и освобождается handle до callback. +Не добавлены новые ссылки/таймеры или зависимость от скорости ответа CM. +Регрессия early cancel прошла вместе с обычной отменой, параллельными потоками, +router restart и destroy при активном server I/O. + +3. Заполнение64 квитанций обрывало массовые мутации + +Прежнее воспроизведение в doc/tasks.md: из76 удалений выполнены64, затем MF_BUSY. +Серверный предел/окно нужны для идемпотентности и остаются: до64 успешных +COMMIT/MKDIR/RENAME/REMOVE, MF_REPLAY_TB=1200000 (120с) от начала обработки. +Квитанции не вытесняются досрочно. Полный кеш возвращает BUSY с cursor=остаток +до ближайшего освобождения в timebase0.1мс; это значение кодируется обычным LE-полем. +Нулевой/недопустимый hint использует обычную задержку5с; недопустимый логируется. + +Клиент этих четырёх команд сохраняет тот же pending/ID/параметры и ожидает +отменяемый retry timer. Остальные BUSY возвращаются callback как раньше. +Общий срок запроса остаётся120с от создания; ни BUSY, ни потеря ответа его не продлевают. +Перед повторной отправкой pump также проверяет срок. Сервер начинает обработку +не раньше создания клиентского запроса, поэтому допустимые повторы не выходят +за окно первой успешной квитанции. Ошибка/timeout не обещают отката серверной мутации. +При постоянной конкуренции запрос может завершиться TIMEOUT; unbounded waiting нет. + +Вариант с продлением срока клиента отклонён при архитектурной сверке до итогового +прогона: поздний BUSY и потеря результата могли вывести повтор за жизнь квитанции. +Один секундный tick также недостаточен для точного освобождения; введён отдельный +retry timer, который снимается до free/cancel/quiesce, с tick fallback при OOM. +Формат125-байтного заголовка не менялся; смысл mutation BUSY.cursor описан в protocol.h. + +Новая сетевая регрессия проходит настоящий клиентский API/router/storage: +-76 MKDIR, включая исчерпание кеша; QUOTA подтверждает ровно76 каталогов. +-Первый успешный ответ намеренно отброшен до клиентского handler; повтор с прежним + ID возвращает успех, не создавая второй объект и не выдавая EXISTS. +-76 STAT/REMOVE, включая новое исчерпание квитанций; QUOTA подтверждает0 объектов/байт/резерва. +Qt recursiveDeletion(server-busy) вызывает storage напрямую через stub bridge: +его сырой MF_BUSY остаётся правильным серверным ответом. Поведение полного +клиентского ожидания и массового удаления проверено C-сетевой регрессией, не этим stub. + +4. FIFO блокировал общий disk worker до проверки типа + +POSIX mf_disk_local_open(source,0) использовал блокирующий open. FIFO без writer +зависал ещё до mf_disk_info, который должен отвергнуть не-regular источник. +Регрессия запускает отдельный процесс до создания pthread, alarm2с ограничивает +только диагностический child. До фикса он завершился SIGALRM (status14), failures=1: +/tmp/member-files-fifo-repro.log. Добавлен O_NONBLOCK, как у серверного open: +FIFO открывается без ожидания, проверка типа даёт INVALID_PATH, fd закрывается. +Обычный и sanitizer-прогоны: child status0, failures0; invalid mode/links имеют WARN. +Произвольный уже выполняющийся системный I/O/хеширование всё ещё не прерывается cancel. + +Сверка архитектуры и описаний + +-Uasync владеет запросами, streams, local FIFO, ACL, CM handles, waiter/pump/timers. + Единственный worker выполняет файловое I/O последовательно; work FIFO, порядок + done при синхронном drain не обещается. Данные живут до completion. +-Обычных работ до64; заранее зарезервированные cleanup jobs обходят лимит. + Quiesce запрещает producers; drain завершает I/O и cleanup, затем storage destroy + выполняется в worker, после него join/free, пока ядро/группы ещё живы. +-Storage принимает только уже проверенные членство/роли. Сам проверяет UID мутации + и токены temp. Квота общая для UID во всех группах; обычные операции не делают rescan. +-Server COMMIT и local download имеют разные гарантии: локальный POSIX rename + не синхронизирует каталог. Ошибка server sync после rename не откатывает публикацию. +-Пути disk API доверенные и уже проверены protocol/storage. Callback LIST получает + заимствованные name/info; порядок файловой системы, не сортировка. Source/target + aliases локальных download исключает caller; одинаковая строка local запрещена. +-Codec проверяет формат и размеры, не ACL и не UTF-8 policy пути. Ошибка decode + может изменить output; encode не проверяет семантику команды. Эти границы описаны. +-Все поля/функции шести .h получили соответствующие контракты и комментарии. + +Проверки и границы + +./build.sh --no-gui -j4 PASS: /tmp/member-files-audit-final-build.log. +Профиль7/7 PASS: member_files, member_files_network, services, encrypted_attachment, +dm_media, attachment_send, member_adapter. Результат: /tmp/member-files-audit-profile-results.log, +подробности: /tmp/member-files-audit-profile-logs/. + +ASAN/UBSan/LSan4/4 PASS, detect_leaks=1, halt_on_error=1: member_files, +member_files_network, services, encrypted_attachment. Отдельная копия: +/home/user/utun4/.cache/member-files-asan-b87xwg24, все встроенные зависимости +инструментированы через CPPFLAGS; системные OpenSSL/FFmpeg не пересобирались. +Логи: /tmp/member-files-sanitizer-build.log, /tmp/member-files-sanitizer-tests-build.log, +/tmp/member-files-sanitizer-results.log, /tmp/member-files-sanitizer-test_*.log. +После этого прогона добавлены только уточнения описаний и WARN без изменения логики. + +В sanitizer-сборке остаётся прежнее предупреждение snprintf в disk_parent_open: +strlen(component)<=MF_NAME_MAX проверяется до записи в leaf[MF_NAME_MAX+1]; +воспроизведённого усечения здесь нет. Остальные предупреждения сборки относятся +к существующему коду других модулей/зависимостей, не скрывались фильтрами компилятора. + +Первые диагностические прогоны до repro5 недействительны для temp: тест был +собран с новым utun_instance.h и старой libutun.a, в startup читались0xaaaaaaaa. +Общая сборка устранила несовпадение; ненужное изменение маршрутного условия отменено. +Один промежуточный старт остановился из-за полного /tmp; очищены только +воспроизводимые объекты предыдущей sanitizer-копии, исходники/журналы сохранены. + +Новый полный make check, Windows/FreeBSD runtime и GUI/Android проверки не выполнялись. +Четыре подтверждённых дефекта закрыты; новых подтверждённых незакрытых в этом объёме нет. diff --git a/doc/tasks.md b/doc/tasks.md index 01b8ed78..1d6618c4 100644 --- a/doc/tasks.md +++ b/doc/tasks.md @@ -4,16 +4,16 @@ ## Текущие задачи -[ ] **member_files: cleanup удаляет чужой temp при отказе создания загрузки** — воспроизведено 24.10.2026 на a198cfb9. - test_member_files_network падает в фазе23: локальный O_EXCL open возвращает MF_EXISTS, - затем TRANSFER_FINISH без проверки владения вызывает mf_disk_local_remove(t->temp). - Прежнее назначение37байт сохраняется, заранее созданный чужой .part исчезает. - Проверены общий check.sh и отдельный повтор: одинаковый FAIL preservation case=0, target=1, foreign_temp=0; - отсутствие .part дополнительно проверено в обоих оставшихся тестовых каталогах. - Нужен учёт успешного создания собственного temp, включая отмену до PREPARE и отказ max_bytes; - проверять/удалять можно только созданный этой передачей файл. Код этого модуля в ETCP-ветке не менялся. - Логи: /tmp/etcp-solo-native-full-check.log, /tmp/etcp-solo-member-network-repeat.log; - подробный первый лог: /tmp/etcp-solo-member-network-suite-failed.log. +[+] **member_files: сверить контракты и исправить ошибки владения/повторов** — завершено 24.10.2026. + Прочитаны все6 реализаций и6 заголовков C-модуля. Исправлены удаление чужого temp, + повторный запуск после раннего cancel с UAF, обрыв массовых мутаций при64 квитанциях + и блокирующий open FIFO до проверки типа. Заголовки и серверная/клиентская документация согласованы. + Подтверждены сохранность файлов при трёх отказах, 76 созданий/76 удалений и потеря успешного ответа без повторной мутации. + Сборка PASS; профиль7/7, ASAN/UBSan/LSan4/4 PASS с detect_leaks=1. Общий GUI/avatar документ остаётся [~]. + Отчёт и границы: [member_files_audit_findings.txt](member_files_audit_findings.txt). + Логи: /tmp/member-files-audit-profile-results.log, /tmp/member-files-sanitizer-results.log. + Прежний FAIL удаления чужого temp из ETCP-проверки закрыт коммитом0f42f92f; + защита от возобновления после раннего cancel —20b4975d. Повторный фикс не требуется. [+] **media_async/media_delivery: завершить сверку контрактов и исправлений** — завершено 24.10.2026. Проверены все11 заголовков двух каталогов по полным реализациям; исправлены alias файлов, @@ -113,6 +113,9 @@ Windows/FreeBSD runtime не проверялся. [ ] **ETCP: полезная утилизация нескольких каналов и доставка при зависании части линков** — общая задача открыта. + Небольшой недобор до условного порога85% сохраняется как результат измерений; + сам по себе не оправдывает изменения/усложнения BBR. Далее по приоритету — STCP backpressure, + смена RTT нескольких путей/BW и повторный отказ; строгие критерии — доставка, паузы и рост очередей. После push исследован одиночный дуплекс2Мбит/с на изолированной базе a198cfb9. Точные7763 ACK показывают задержку обратной очереди: пример DATA15.909мс, hold0.029мс, обратный ACK32.219мс; DOWN продолжает ждать flight7644 при цели5915байт. @@ -805,13 +808,16 @@ нескольких ретрансляторов, нескольких автоматических источников/групп и mobile standby на устройстве (doc/stream.txt, раздел 16). Сохранённые проверки отдельных этапов не закрывают всю матрицу. -[ ] **Member files: исчерпание кеша результатов массовых изменений** — 10.10.2026: +[+] **Member files: исчерпание кеша результатов массовых изменений** — исправлено 24.10.2026. Воспроизведено test_member_files_browser::recursiveDeletion(server-busy): из 76 объектов удаляются 64, следующая команда получает MF_BUSY, остаются 12. storage.c: STORE_RECEIPTS=64, RECEIPT_TB=1200000 (120 секунд). Лимит общий для успешных COMMIT/MKDIR/RENAME/REMOVE; затрагивает также рекурсивное копирование. - Нужен отдельный этап управления временем жизни receipts с сохранением идемпотентности повторных запросов; - простое вытеснение старых записей нарушает гарантию повторного ответа. После исправления изменить регрессию - на успешное удаление более 64 объектов и проверить повторные запросы/потерю ответа. + Сохранены64 квитанции/120с без досрочного вытеснения. BUSY.cursor сообщает ближайшее освобождение, + клиент этих мутаций ждёт отменяемый таймер с прежним ID/параметрами и общим сроком120с без продления. + Новая сетевая регрессия через настоящий клиентский API:76 созданий, потеря успешного ответа, + затем76 удалений; QUOTA подтверждает76 и0 каталогов без дубликатов. Обычный и sanitizer-прогоны PASS. + Старый Qt-stub вызывает storage напрямую и по-прежнему видит допустимый сырой MF_BUSY; + он не проверяет новый клиентский retry. Подробности: [member_files_audit_findings.txt](member_files_audit_findings.txt). [ ] **Desktop: провалы вероятности VAD при компрессии** — диагностика 09.10.2026: Воспроизведение пользователя: длительная гласная, затем временная нечувствительность; ближе к концу лога несколько повторов. diff --git a/src/member_files/disk.c b/src/member_files/disk.c index 7a7d13f7..02e7e230 100644 --- a/src/member_files/disk.c +++ b/src/member_files/disk.c @@ -58,7 +58,10 @@ static int disk_win_error(const char* operation) { static int disk_handle_info(HANDLE h, struct mf_disk_info* out) { BY_HANDLE_FILE_INFORMATION i; if (!GetFileInformationByHandle(h, &i)) return disk_win_error("GetFileInformationByHandle"); - if ((i.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) || i.nNumberOfLinks > 1) return MF_INVALID_PATH; + if ((i.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) || i.nNumberOfLinks > 1) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: unsupported file attributes=%lu links=%lu", i.dwFileAttributes, i.nNumberOfLinks); + return MF_INVALID_PATH; + } memset(out, 0, sizeof(*out)); out->type = i.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY ? MF_DIRECTORY : MF_FILE; out->size = out->type == MF_FILE ? ((uint64_t)i.nFileSizeHigh << 32) | i.nFileSizeLow : 0; @@ -254,7 +257,11 @@ static int disk_parent_open(struct mf_disk_root* root, const char* path, struct } static int disk_posix_info(const struct stat* st, struct mf_disk_info* i) { - if ((!S_ISREG(st->st_mode) && !S_ISDIR(st->st_mode)) || (S_ISREG(st->st_mode) && st->st_nlink != 1)) return MF_INVALID_PATH; + if ((!S_ISREG(st->st_mode) && !S_ISDIR(st->st_mode)) || (S_ISREG(st->st_mode) && st->st_nlink != 1)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: unsupported file mode=%lo links=%llu", + (unsigned long)st->st_mode, (unsigned long long)st->st_nlink); + return MF_INVALID_PATH; + } memset(i, 0, sizeof(*i)); i->type = S_ISDIR(st->st_mode) ? MF_DIRECTORY : MF_FILE; i->size = i->type == MF_FILE ? (uint64_t)st->st_size : 0; i->mtime = (uint64_t)st->st_mtime; i->identity[0] = st->st_dev; i->identity[1] = st->st_ino; @@ -352,7 +359,7 @@ int mf_disk_resize(int fd, uint64_t size) { return ftruncate(fd, (off_t)size) ? int mf_disk_sync(int fd) { return fsync(fd) ? disk_error("sync", NULL) : MF_OK; } int mf_disk_close(int fd) { return close(fd) ? disk_error("close", NULL) : MF_OK; } int mf_disk_local_open(const char* path, int create, int* fd) { - *fd = open(path, O_CLOEXEC | O_NOFOLLOW | (create ? O_RDWR | O_CREAT | O_EXCL : O_RDONLY), 0600); + *fd = open(path, O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK | (create ? O_RDWR | O_CREAT | O_EXCL : O_RDONLY), 0600); return *fd < 0 ? disk_error("local open", path) : MF_OK; } int mf_disk_local_publish(const char* from, const char* to) { return rename(from, to) ? disk_error("local publish", to) : MF_OK; } diff --git a/src/member_files/disk.h b/src/member_files/disk.h index bbdaa334..49dfc5a0 100644 --- a/src/member_files/disk.h +++ b/src/member_files/disk.h @@ -1,39 +1,64 @@ -/* Примитивы диска для storage и клиентского worker. На POSIX пути разрешаются +/* Синхронные примитивы диска для storage и клиентского worker; не вызывать из uasync callbacks. + * Возврат — MF_OK либо положительный mf_status, не errno/-1. Аргументы/выходы обязательны, кроме отмеченных NULL. + * Серверные path/from/to — уже проверенные относительные пути; сам слой не запрещает '.'/'..'. + * На POSIX пути разрешаются * относительно открытого каталога через *at/O_NOFOLLOW. Windows удерживает каталоги * без FILE_SHARE_DELETE и отвергает reparse points; имена преобразуются UTF-8→UTF-16. - * fd принадлежит вызывающему; read/write работают по 64-битному смещению. */ + * fd принадлежит вызывающему; read/write работают по 64-битному смещению. + * Локальные пути приложения не ограничены корнем storage; их родители/назначение остаются ответственностью caller. */ #ifndef MEMBER_FILES_DISK_H #define MEMBER_FILES_DISK_H #include "protocol.h" #include "../../lib/platform_compat.h" -struct mf_disk_info { uint64_t size, mtime, identity[6]; uint8_t type; }; +struct mf_disk_info { + uint64_t size; // Байты regular-файла; каталог — 0. + uint64_t mtime; // Unix seconds; для version используются более точные identity. + uint64_t identity[6]; // Непрозрачный снимок inode/file ID и времён; платформенно зависим, не wire. + uint8_t type; // MF_FILE/MF_DIRECTORY; другие типы и regular hardlink отклоняются. +}; struct mf_disk_root { - char path[2048]; + char path[2048]; // Заимствованный при open путь копируется; предел2047 байт + NUL. #ifdef _WIN32 - HANDLE handle; + HANDLE handle; // Собственный handle корня, закрывается после всех операций. #else - int fd; + int fd; // Собственный directory fd; операции разрешают имена относительно него. #endif }; +/* Создать корень, если отсутствует (не создаёт родителей), и закрепить его. Close нужен только после успеха. */ int mf_disk_root_open(struct mf_disk_root* root, const char* path); void mf_disk_root_close(struct mf_disk_root* root); +/* Открыть regular-файл без усечения. create!=0 — эксклюзивное создание, writable!=0 — read/write. + * fd использовать/закрывать только после MF_OK; отказ не гарантирует значение *fd. */ int mf_disk_open(struct mf_disk_root* root, const char* path, int create, int writable, int* fd); +/* Снимок без чтения содержимого: stat — по имени, info — уже открытого fd. Выход использовать только при успехе. */ int mf_disk_stat(struct mf_disk_root* root, const char* path, struct mf_disk_info* info); int mf_disk_info(int fd, struct mf_disk_info* info); -/* created, если не NULL, показывает факт создания даже при последующей ошибке sync. */ +/* Создать один каталог. created, если не NULL, показывает факт создания даже при последующей ошибке sync. */ int mf_disk_mkdir(struct mf_disk_root* root, const char* path, int* created); +/* Удалить файл либо пустой каталог. Ошибка sync на POSIX возможна уже после удаления. */ int mf_disk_remove(struct mf_disk_root* root, const char* path, int directory); +/* Заменяющий rename: запрет замены при обычном RENAME проверяет storage. На POSIX sync обоих родителей; + * ошибка sync не означает откат уже выполненного rename. */ int mf_disk_rename(struct mf_disk_root* root, const char* from, const char* to); +/* name/info заимствованы на время callback; 0 — продолжить, >0 — успешная ранняя остановка, <0 — -mf_status. */ typedef int (*mf_disk_entry_fn)(void* arg, const char* name, const struct mf_disk_info* info); +/* Перечислить записи в порядке файловой системы, без '.'/'..'. Пустой path означает корень. */ int mf_disk_list(struct mf_disk_root* root, const char* path, mf_disk_entry_fn cb, void* arg); +/* Точный диапазон: 0 байт допустимо, короткий read — ошибка. Caller гарантирует offset+size <= INT64_MAX. */ int mf_disk_read(int fd, uint64_t offset, void* data, size_t size); int mf_disk_write(int fd, uint64_t offset, const void* data, size_t size); +/* Изменить длину (size<=INT64_MAX), синхронизировать содержимое либо закрыть fd; каждый вызов независим. */ int mf_disk_resize(int fd, uint64_t size); int mf_disk_sync(int fd); int mf_disk_close(int fd); +/* Локальный source (create=0) либо эксклюзивный read/write temp (create!=0), без усечения. + * Тип проверяет caller через info; POSIX O_NONBLOCK позволяет отвергнуть FIFO до блокирующего чтения. */ int mf_disk_local_open(const char* path, int create, int* fd); +/* Заменить local назначение: POSIX rename без sync каталога; Windows MOVEFILE_WRITE_THROUGH. */ int mf_disk_local_publish(const char* from, const char* to); +/* Удалить собственный local temp; отсутствие самого файла допустимо. Владение проверяет caller. */ int mf_disk_local_remove(const char* path); +/* SHA256 ровно size байт от нуля, включая пустой файл; ошибка делает hash непригодным. fd остаётся у caller. */ int mf_disk_hash(int fd, uint64_t size, uint8_t hash[32]); #endif diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 18a15074..35cc1ad5 100644 --- a/src/member_files/member_files.c +++ b/src/member_files/member_files.c @@ -16,16 +16,17 @@ #define MF_MAX_PENDING 16 #define MF_MAX_JOBS 32 -#define MF_REQUEST_TB 1200000ULL struct mf_pending { struct mf_pending* next; struct member_files* files; - uint64_t group, server, started, sent; + uint64_t group, server, started, sent, retry_at; + int busy; // Уже залогировано ожидание квитанции; общий срок запроса не продлевается. struct mf_message request; mf_response_fn callback; void* arg; void* pump; + void* retry; struct CONN_MGR_HANDLE* cm; struct queue_waiter_handle waiter; }; @@ -170,6 +171,7 @@ int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const struct static void mf_pending_finish(struct mf_pending* p, const struct mf_message* response, int status) { struct member_files* f = p->files; if (p->pump) uasync_call_soon_cancel(f->inst->ua, p->pump); + if (p->retry) uasync_cancel_timeout(f->inst->ua, p->retry); etcp_router_cancel_send_ready(f->inst, p->group, p->server, ETCP_RT_ID_MEMBER_FILES, &p->waiter); if (p->cm) conn_mgr_close(p->cm); struct mf_pending** link = &f->pending; @@ -188,20 +190,32 @@ static void mf_pending_finish(struct mf_pending* p, const struct mf_message* res static void mf_pending_pump(void* arg) { struct mf_pending* p = arg; p->pump = NULL; if (p->files->closing) return; + if (p->retry) { uasync_cancel_timeout(p->files->inst->ua, p->retry); p->retry = NULL; } + if (get_time_tb() - p->started >= MF_REPLAY_TB) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: request expired before send id=%llu cmd=%s", + (unsigned long long)p->request.request_id, mf_command_text(p->request.command)); + mf_pending_finish(p, NULL, MF_TIMEOUT); return; + } int rc = mf_send(p->files, p->group, p->server, &p->request); - if (!rc) p->sent = get_time_tb(); + if (!rc) { p->sent = get_time_tb(); p->retry_at = 0; } else if (rc > 0) etcp_router_on_send_ready(p->files->inst, p->group, p->server, ETCP_RT_ID_MEMBER_FILES, &p->waiter, mf_client_ready, p); } static void mf_pending_wake(struct mf_pending* p) { - if (p->pump || p->files->closing) return; + if (p->pump || p->files->closing || get_time_tb() < p->retry_at) return; p->pump = uasync_call_soon(p->files->inst->ua, p, mf_pending_pump); if (!p->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: client pump allocation failed"); } static void mf_client_ready(struct ll_queue* q, void* arg) { (void)q; mf_pending_wake(arg); } +/* Отдельный таймер не округляет срок освобождения квитанции до секундного sweep. */ +static void mf_pending_retry(void* arg) { + struct mf_pending* p = arg; p->retry = NULL; p->retry_at = 0; + mf_pending_wake(p); +} + static void mf_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t group, enum conn_mgr_event event, void* arg) { (void)h; (void)peer; (void)group; struct mf_pending* p = arg; @@ -385,18 +399,18 @@ static void mf_tick(void* arg) { mf_receive_resume(f); for (;;) { struct mf_pending* expired = NULL; - for (struct mf_pending* p = f->pending; p; p = p->next) if (now - p->started >= MF_REQUEST_TB) { expired = p; break; } + for (struct mf_pending* p = f->pending; p; p = p->next) if (now - p->started >= MF_REPLAY_TB) { expired = p; break; } if (!expired) break; DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: timeout id=%llu cmd=%u", (unsigned long long)expired->request.request_id, expired->request.command); mf_pending_finish(expired, NULL, MF_TIMEOUT); } for (struct mf_pending* p = f->pending; p; p = p->next) - if (!p->sent || now - p->sent >= 50000) mf_pending_wake(p); + if (p->retry_at ? now >= p->retry_at : !p->sent || now - p->sent >= 50000) mf_pending_wake(p); struct mf_job* j = f->jobs; while (j) { struct mf_job* next = j->next; - if (j != f->worker && j->created && now - j->created >= MF_REQUEST_TB) { + if (j != f->worker && j->created && now - j->created >= MF_REPLAY_TB) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: server reply/queue timeout peer=%llu", (unsigned long long)j->peer); mf_job_free(j); } j = next; @@ -427,6 +441,26 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: mismatched response id=%llu", (unsigned long long)m.request_id); return; } if (!member_files_member(f->inst, group, peer, 1)) { m.status = MF_ACCESS_DENIED; m.data_len = 0; } + if (m.status == MF_BUSY && (m.command == MF_COMMIT || (m.command >= MF_MKDIR && m.command <= MF_REMOVE_PATH))) { + if (m.cursor > MF_REPLAY_TB) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: invalid busy hint id=%llu delay_tb=%llu; using5s", + (unsigned long long)m.request_id, (unsigned long long)m.cursor); + uint64_t now = get_time_tb(), delay = m.cursor && m.cursor <= MF_REPLAY_TB ? m.cursor : 50000; + p->retry_at = now + delay; + p->sent = now; + if (p->pump) { uasync_call_soon_cancel(f->inst->ua, p->pump); p->pump = NULL; } + if (p->retry) uasync_cancel_timeout(f->inst->ua, p->retry); + p->retry = uasync_set_timeout(f->inst->ua, (int)delay, p, mf_pending_retry, "member_files_retry"); + if (!p->retry) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: busy retry timer failed id=%llu; tick will retry", + (unsigned long long)m.request_id); + etcp_router_cancel_send_ready(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES, &p->waiter); + if (!p->busy) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: mutation busy; retaining request id=%llu cmd=%s limit=120s", + (unsigned long long)m.request_id, mf_command_text(m.command)); + p->busy = 1; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: busy retry id=%llu elapsed_ms=%llu delay_tb=%llu retry_at=%llu", + (unsigned long long)m.request_id, (unsigned long long)((now - p->started) / 10), + (unsigned long long)delay, (unsigned long long)p->retry_at); + return; + } mf_pending_finish(p, &m, 0); break; } return; diff --git a/src/member_files/member_files.h b/src/member_files/member_files.h index e00d6e03..909e862c 100644 --- a/src/member_files/member_files.h +++ b/src/member_files/member_files.h @@ -17,7 +17,8 @@ typedef void (*mf_response_fn)(void* arg, const struct mf_message* response); // ── Жизненный цикл модуля ── -// Создать состояние сервера/клиента и запустить дисковый worker; 0/-1. +// Создать состояние сервера/клиента и worker; 0/-1. Требует живые ua/config/router/chat/SQLite, повторный init — ошибка. +// Стартовый scan асинхронный: init=0 ещё не означает готовность локального storage; remote client остаётся доступен. int member_files_init(struct UTUN_INSTANCE* inst); // Запретить новые команды и начать отмену запросов/передач; запущенный дисковый I/O завершается отдельно. @@ -26,17 +27,20 @@ void member_files_quiesce(struct UTUN_INSTANCE* inst); // Дождаться дискового worker и освободить контекст; вне callbacks, до удаления CHAT-групп. void member_files_destroy(struct UTUN_INSTANCE* inst); -// Отменить запросы и передачи одной группы перед её удалением. +// Отменить свои запросы/передачи одной группы перед её удалением; дерево storage этим не удаляется. void member_files_cancel_group(struct UTUN_INSTANCE* inst, uint64_t group); // ── Команды файлового сервера ── /* Отправить одну команду; request_id=0 выбирает случайный ID. NULL — отказ запуска, callback не вызывается. - * При успехе callback ровно один, включая отмену; handle освобождён до callback, arg принадлежит вызывающему. */ + * При успехе callback ровно один, включая отмену; handle освобождён до callback, arg принадлежит вызывающему. + * Общий timeout120с. BUSY для COMMIT/MKDIR/RENAME/REMOVE сохраняет ID и ждёт подсказанный сервером срок + * освобождения квитанции (иначе5с), не продлевая общий timeout. Остальные BUSY передаются callback. */ struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t server, const struct mf_message* request, mf_response_fn callback, void* arg); // Синхронно завершить свой запрос ответом MF_CANCELLED; handle после вызова недействителен, NULL допустим. +// Это отмена локального ожидания; уже запущенная серверная мутация не откатывается. void member_files_cancel(struct mf_pending* pending); // Отменить pending-запрос с этим request_id; не отменяет потоковую передачу файла. @@ -44,14 +48,20 @@ void member_files_cancel_request_id(struct UTUN_INSTANCE* inst, uint64_t id); // ── Проверка доступа ── -// 1 — действующий мембер; storage!=0 дополнительно требует подписанную роль хранителя. +// 1 — полная source=0/deleted=0 запись мембера в локальной CHAT-таблице, иначе0. +// storage!=0 требует adm_tags.storage=yes и сохранённую64-байтную подпись; peers.storage не даёт полномочий. int member_files_member(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node, int storage); // ── Команда интерфейса ── -struct mf_gui_request { struct UTUN_INSTANCE* inst; uint64_t group, server; struct mf_message message; }; +struct mf_gui_request { + struct UTUN_INSTANCE* inst; // Заимствованное ядро до ответа. + uint64_t group, server; // CHAT-группа и явно выбранный storage, в том числе self. + struct mf_message message; // Команда; для корреляции GUI выбирает ненулевой request_id. +}; -// arg — u_malloc/u_calloc mf_gui_request; освобождается после ответа CHAT_EVT_MEMBER_FILES_RESPONSE. +// arg — u_malloc/u_calloc mf_gui_request; владение переходит при вызове trampoline, освобождается после ответа. +// CHAT_EVT_MEMBER_FILES_RESPONSE: group:8LE/server:8LE, затем mf_encode(response). Отказ запуска даёт ACCESS_DENIED. void member_files_gui_request(void* arg); // ── Потоковая передача файлов ── @@ -63,12 +73,17 @@ struct mf_transfer_request { // download: по ответу STAT, до открытия локального temp и передачи данных. uint64_t max_bytes; int upload; // ненулевой — загрузить local на сервер, 0 — скачать remote - uint8_t version[16]; // версия из STAT; нули при создании нового серверного файла + uint8_t version[16]; // Upload:ожидаемая версия/нули для создания; download:ожидаемая версия/нули для текущей. char remote[MF_PATH_MAX + 1], local[MF_PATH_MAX + 1]; // путь в каталоге owner и локальный путь приложения }; struct mf_transfer; -struct mf_transfer_progress { uint64_t id, done, total; uint16_t status; uint8_t finished; }; +struct mf_transfer_progress { + uint64_t id; // Фактический ID handle, включая автоматически выбранный. + uint64_t done, total; // Upload:принятые router байты; download:записанные на диск, до финального sync/publish. + uint16_t status; // mf_status; результат окончательный только при finished. + uint8_t finished; // 0 progress (может отсутствовать),1 ровно одно завершение. +}; typedef void (*mf_transfer_fn)(void* arg, const struct mf_transfer_progress* progress); // Один handle на файл; до 16 клиентских передач одновременно, включая upload/download к одному пиру. @@ -76,9 +91,13 @@ typedef void (*mf_transfer_fn)(void* arg, const struct mf_transfer_progress* pro // finished приходит ровно один раз после закрытия/публикации; handle уже освобождён до callback. // Прогресс заимствован на время callback. Cancel асинхронно ждёт свой I/O и не затрагивает другие handles. // Request копируется; исходную структуру можно освободить сразу, arg должен жить до finished. +// Download заменяет local только после проверки/закрытия своего temp; отказ/отмена до создания не удаляет чужой temp. +// Совпадение строк local двух download запрещено; aliases/подмену локальных файлов исключает caller. +// Upload требует стабильный regular source; изменение снимка даёт STALE. Пустой файл допустим. struct mf_transfer* member_files_transfer_start(const struct mf_transfer_request* request, mf_transfer_fn callback, void* arg); // Запросить отмену своей передачи; завершение приходит через finished после закрытия её файлового I/O. +// Допустимо сразу после start: поздний CM UP не запускает подготовку/STAT заново. Финализируемый результат уже не отменяется. void member_files_transfer_cancel(struct mf_transfer* transfer); // ── Команды интерфейса для передачи и отмены ── @@ -87,7 +106,7 @@ void member_files_transfer_cancel(struct mf_transfer* transfer); * state=1 progress, 2 finished. arg — u_malloc/u_calloc mf_transfer_request, освобождается после finished. */ void member_files_gui_transfer(void* arg); -struct mf_cancel_request { struct UTUN_INSTANCE* inst; uint64_t id; }; +struct mf_cancel_request { struct UTUN_INSTANCE* inst; uint64_t id; }; // Заимствованное ядро и фактический ID команды/передачи. // arg — u_malloc/u_calloc mf_cancel_request, освобождается здесь; отменяет передачу либо одиночный запрос по id. void member_files_gui_cancel(void* arg); diff --git a/src/member_files/private.h b/src/member_files/private.h index b1adb372..5529e695 100644 --- a/src/member_files/private.h +++ b/src/member_files/private.h @@ -1,4 +1,6 @@ -/* Внутреннее состояние owner, только для member_files/; storage принадлежит единственному worker. */ +/* Внутреннее состояние member_files/, все списки/таймеры принадлежат uasync. + * Disk worker последовательно владеет storage и своими снимками I/O; completion возвращает управление uasync. + * Start: worker/reserved cleanup → storage → router/tick → initialize. Stop: quiesce → drain → storage cleanup → join/free. */ #ifndef MEMBER_FILES_PRIVATE_H #define MEMBER_FILES_PRIVATE_H #include "member_files.h" @@ -9,31 +11,41 @@ struct mf_job; struct mf_transfer; struct mf_local; struct member_files { - struct UTUN_INSTANCE* inst; - struct mf_storage* storage; - struct mf_worker* disk; - struct mf_work* cleanup; - struct mf_pending* pending; - struct mf_job* jobs; - struct mf_job* worker; - struct mf_transfer* transfers; - struct mf_local* locals; - void* tick; - void* pump; - unsigned pending_count, job_count, transfer_count; - unsigned server_transfer_count; - unsigned local_count; - int closing, used_storage, sweep_due; - int storage_status, init_result; + struct UTUN_INSTANCE* inst; // Заимствованное ядро; ua/router/группы живут до destroy. + struct mf_storage* storage; // Собственный storage; освобождается worker после всех stream cleanup. + struct mf_worker* disk; // Единственный FIFO дискового I/O; uasync владеет handle. + struct mf_work* cleanup; // Резерв освобождения storage, создан до начала I/O. + struct mf_pending* pending; // Клиентские команды, до16; handle удаляется до callback. + struct mf_job* jobs; // До32 серверных queued/worker/reply задач, включая sweep. + struct mf_job* worker; // Текущая серверная задача; не включает startup/клиентские stream I/O. + struct mf_transfer* transfers; // Собственные client/server streams и их отложенная очистка. + struct mf_local* locals; // До32 сообщений self→self, общий FIFO запросов/ответов. + void* tick; // Таймер1с: pending/stream timeout, повторы, sweep/resume. + void* pump; // Отменяемый call_soon серверного pump, не worker. + unsigned pending_count, job_count, transfer_count; // Числа pending/server jobs/client streams. + unsigned server_transfer_count; // Серверные streams, независимо от client limit16. + unsigned local_count; // Длина локального FIFO. + int closing, used_storage, sweep_due; // Quiesce, необходимость периодического sweep и ожидающий sweep. + int storage_status; // Uasync:BUSY до startup completion, затем MF_OK либо постоянная ошибка. + int init_result; // Worker пишет результат initialize; читать только после completion/drain. }; +/* Запросить остановку streams группы (0 — всех), дождаться их I/O через worker drain отдельно. */ void mf_transfers_quiesce(struct member_files* files, uint64_t group); +/* 1 — DATA помещается в кольцо либо относится к уже неактивному потоку;0 — router должен удержать пакет. */ int mf_transfer_receive_ready(struct member_files* files, uint64_t group, uint64_t peer, int kind, uint64_t id); +/* Обработать проверенный STREAM-пакет; group/peer — end-to-end identity router. */ void mf_transfer_dispatch(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); +/* Завершить потоки пары с IO_ERROR при reset/close router; прочие пары не затрагиваются. */ void mf_transfers_peer_down(struct member_files* files, uint64_t group, uint64_t peer); +/* Планировать готовые pumps / обработать прогресс и таймауты; все функции вызываются в uasync. */ void mf_transfers_wake(struct member_files* files); void mf_transfers_tick(struct member_files* files, uint64_t now); +/* Отправить signed+encrypted либо поставить local:0 принято,1 backpressure,-1 ошибка. message копируется. */ int mf_send(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); +/* 1/0: enabled, signed storage-role self, действующие peer/owner, запись только своего UID. Без I/O. */ int mf_server_allowed(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); +/* Пробудить заблокированный router receive и голову local FIFO, если появились места. */ void mf_receive_resume(struct member_files* files); +/* Поставить копию ответа с указанным статусом в ограниченную очередь; отказ логируется. */ void mf_reply(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* request, int status); #endif diff --git a/src/member_files/protocol.h b/src/member_files/protocol.h index 45ee9a15..97316817 100644 --- a/src/member_files/protocol.h +++ b/src/member_files/protocol.h @@ -1,9 +1,12 @@ -/* Файловый протокол участников CHAT-группы. Все числа на проводе little-endian. +/* Файловый протокол участников CHAT-группы. Codec не проверяет членство/подпись/шифрование: это делает сервис. + * Все числа на проводе little-endian. * Группа и подписанный отправитель приходят от router; owner — UID каталога. * version — непрозрачный снимок объекта, upload — случайный ID временного файла. * Нулевая version в BEGIN означает создание; замена требует version из STAT. * LIST: cursor — число уже выданных записей; version фиксирует снимок каталога. * READ/WRITE: offset и size задают диапазон; WRITE.data_len обязан равняться size. + * COMMIT/MKDIR/RENAME/REMOVE: RESPONSE BUSY.cursor — срок до освобождения квитанции в0.1мс; + * 0 означает обычный повтор через5с. Общий клиентский срок120с не продлевается. * Данные LIST — последовательность entry: name_len:2,type:1,size:8,mtime:8,version:16,name. * Токены меняются при перезапуске сервера. Никаких ABI-зависимых структур в wire. */ #ifndef MEMBER_FILES_PROTOCOL_H @@ -25,10 +28,13 @@ extern "C" { #define MF_AVATAR_MAX_BYTES (1024 * 1024) #define MF_AVATAR_ORIGINAL_MAX_BYTES (32 * 1024 * 1024) #define MF_AVATAR_MANIFEST_MAX_BYTES 4096 +/* Имя объекта аватарки:64 lowercase hex и .png/.jpg/.webp/.json; вход — NUL-строка, NULL даёт0. + * valid/manifest_valid возвращают1/0; file_limit — лимит32MiB для изображения,4096 для JSON,0 при неверном имени. */ int mf_avatar_file_valid(const char* file); int mf_avatar_manifest_valid(const char* file); uint64_t mf_avatar_file_limit(const char* file); #define MF_IDLE_TB 100000ULL /* 10 секунд */ +#define MF_REPLAY_TB 1200000ULL /* Окно повторов запроса/успешной квитанции:120с. */ #define MF_REQUEST 1 #define MF_RESPONSE 2 #define MF_FILE 1 @@ -45,22 +51,38 @@ enum mf_status { }; struct mf_message { - uint8_t kind, command, type, more; - uint16_t status; - uint64_t request_id, owner, offset, size, cursor, mtime; - uint8_t version[16], upload[16], hash[32]; - char path[MF_PATH_MAX + 1], target[MF_PATH_MAX + 1]; - uint32_t data_len; - uint8_t data[MF_BLOCK_SIZE]; + uint8_t kind, command; // MF_REQUEST/RESPONSE и mf_command; group/src берутся отдельно из router. + uint8_t type, more; // MF_FILE/DIRECTORY и признак продолжения LIST/READ (0/1). + uint16_t status; // mf_status; у запроса всегда MF_OK. + uint64_t request_id; // Ненулевой ID одной неизменной команды; у DATA/END/CANCEL — ID потока. + uint64_t owner; // UID дерева;0 только для виртуального LIST корня группы. + uint64_t offset, size; // Диапазон READ/WRITE/DATA либо полный размер BEGIN/END/STAT. + uint64_t cursor; // LIST: число выданных записей; открытие STREAM:ID потока; mutation BUSY:retry_after_tb. + uint64_t mtime; // Unix seconds в ответах/записях каталога. + uint8_t version[16]; // Непрозрачный снимок объекта; нули BEGIN означают создание. + uint8_t upload[16]; // Случайный токен temp/потока; проверяется вместе с group/peer/owner/path. + uint8_t hash[32]; // SHA256 полного файла для BEGIN/COMMIT/END. + char path[MF_PATH_MAX + 1], target[MF_PATH_MAX + 1]; // NUL-строки; target используется RENAME. + uint32_t data_len; // Длина data; фактический предел MF_BLOCK_SIZE, wire использует u16. + uint8_t data[MF_BLOCK_SIZE]; // Байты файла, LIST entries или QUOTA, по команде. }; +/* Прочитать/записать ровно8 байт LE; буферы обязательны и имеют достаточный размер. */ uint64_t mf_get64(const uint8_t* p); void mf_put64(uint8_t* p, uint64_t value); +/* 1 — все n байт нулевые, иначе0. */ int mf_zero(const uint8_t* p, size_t n); -/* Проверка UTF-8 и переносимых компонентов. Пустой путь допустим только для корня. */ +/* 1/0: строгий UTF-8, путь<=1024 байт, до60 компонентов по1..255 байт; root_allowed разрешает пустую строку. + * Отвергает leading '.', trailing '.'/' ', ASCII C0/DEL и Windows device names. + * Запрещены обратная косая черта, двоеточие, '*', '?', '"', '<', '>', '|'. */ int mf_path_valid(const char* path, int root_allowed); +/* Записать wire: длина>0 либо-1 при переполнении; выход не меняется при отказе длины. + * Проверяет размеры, но не семантику command/ID/path. message/wire обязательны и не перекрываются. */ int mf_encode(const struct mf_message* message, uint8_t* wire, size_t capacity); +/* 0/-1: проверить формат/точную длину/общие поля и декодировать. При ошибке message может быть изменён. + * Невалидный UTF-8/путь и смысл полей отдельных команд проверяет вызывающий сервис/storage. */ int mf_decode(const uint8_t* wire, size_t length, struct mf_message* message); +/* Заимствованные статические строки; неизвестное значение возвращает Unknown error/UNKNOWN. */ const char* mf_status_text(int status); const char* mf_command_text(int command); #ifdef __cplusplus diff --git a/src/member_files/storage.c b/src/member_files/storage.c index 053016e9..115c3a33 100644 --- a/src/member_files/storage.c +++ b/src/member_files/storage.c @@ -12,7 +12,6 @@ #define STORE_FILES 64 #define STORE_UPLOADS 16 #define STORE_RECEIPTS 64 -#define RECEIPT_TB 1200000ULL #define STORE_USERS 256 // Windows сравнивает имена без регистра; алиас не должен обходить неизменяемость avatars. @@ -578,6 +577,8 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con return; } int mutates = q->command == MF_COMMIT || q->command == MF_MKDIR || q->command == MF_RENAME || q->command == MF_REMOVE_PATH; + uint64_t first_expiry = UINT64_MAX; + if (mutates) r->cursor = 0; struct store_receipt* slot = NULL; uint8_t wire[MF_WIRE_MAX], fingerprint[32]; int n = mf_encode(q, wire, sizeof(wire)); @@ -586,6 +587,7 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con for (unsigned i = 0; i < STORE_RECEIPTS; i++) { struct store_receipt* p = &s->receipts[i]; if (p->until <= now) { if (!slot) slot = p; continue; } + if (p->until < first_expiry) first_expiry = p->until; if (p->group != group || p->peer != peer) continue; if (p->response.request_id == q->request_id) { if (memcmp(p->fingerprint, fingerprint, 32)) r->status = MF_BAD_PACKET; @@ -597,7 +599,11 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con *r = p->response; r->request_id = q->request_id; return; } } - if (mutates && !slot) r->status = MF_BUSY; + if (mutates && !slot) { + r->status = MF_BUSY; r->cursor = first_expiry - now; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: receipts full id=%llu retry_after_tb=%llu", + (unsigned long long)q->request_id, (unsigned long long)r->cursor); + } else { mf_storage_sweep(s, now); r->status = (uint16_t)store_command(s, group, peer, q, r); /* Idle отсчитывается от завершения I/O, а не от начала медленного read/write. */ @@ -615,7 +621,7 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con } } if (!r->status && mutates) { - slot->group = group; slot->peer = peer; slot->until = now + RECEIPT_TB; + slot->group = group; slot->peer = peer; slot->until = now + MF_REPLAY_TB; memcpy(slot->fingerprint, fingerprint, 32); slot->response = *r; } DEBUG_TRACE(DEBUG_CATEGORY_MEDIA, "member_files: storage result id=%llu cmd=%s status=%s bytes=%u more=%u cursor=%llu", diff --git a/src/member_files/storage.h b/src/member_files/storage.h index 4dd95ab3..b0adbee6 100644 --- a/src/member_files/storage.h +++ b/src/member_files/storage.h @@ -1,17 +1,32 @@ -/* Серверное хранилище. Вызывать execute/sweep/destroy последовательно, только без - * одновременно работающего worker. create не обращается к диску; initialize выполняет - * стартовый обход в worker до запросов. ACL проверяет member_files.c перед каждой работой. Файлы временных загрузок +/* Серверное хранилище. После create все initialize/execute/sweep/destroy выполняются + * последовательно одним disk worker; блокировки внутри storage нет. create не обращается к диску; + * ещё не инициализированный пустой owner допустимо destroy в потоке создателя. + * initialize выполняет стартовый обход до запросов. ACL проверяет member_files.c перед каждой работой. + * Storage сам проверяет group/peer/owner и принадлежность мутации UID, но не членство/подписанные роли. + * Файлы временных загрузок * закрываются после 10 секунд простоя, удаляются после upload_ttl; restart отменяет их. - * Открытые regular-файлы кешируются, максимум 64. Родительские пути не следуют ссылкам. */ + * Открытые regular-файлы кешируются, максимум64 (включая temp); активных upload до16. + * Квота/числа объектов общие для UID во всех группах и восстанавливаются только при initialize. + * Успешные COMMIT/MKDIR/RENAME/REMOVE занимают одну из64 квитанций на MF_REPLAY_TB от начала обработки. + * Родительские пути не следуют ссылкам. Файлы и счётчики не образуют откатываемую транзакцию. */ #ifndef MEMBER_FILES_STORAGE_H #define MEMBER_FILES_STORAGE_H #include "protocol.h" struct mf_storage; +/* root — NUL-путь1..2047 байт, quota/TTL>0 (байты и timebase0.1мс); копирует root, возвращает owner/NULL. */ struct mf_storage* mf_storage_create(const char* root, uint64_t quota, uint64_t upload_ttl_tb); -/* Однократный стартовый обход. До его завершения execute возвращает BUSY, после ошибки — её код. */ +/* MF_OK/положительный mf_status: создать/открыть root, удалить старые .uploads, восстановить квоты. + * Повтор после успеха — no-op; после ошибки возвращает тот же код, нужен новый owner/restart. + * До завершения execute возвращает BUSY, после ошибки — её код, частичные счётчики не выдаются. */ int mf_storage_initialize(struct mf_storage* storage); +/* Закрыть fd/корень, удалить активные temp и освободить owner; опубликованные файлы сохраняются, NULL допустим. */ void mf_storage_destroy(struct mf_storage* storage); +/* Закрыть idle fd и удалить просроченные uploads; now — монотонный timebase0.1мс, NULL допустим. */ void mf_storage_sweep(struct mf_storage* storage, uint64_t now); +/* Выполнить LIST..REMOVE либо QUOTA; STREAM преобразует transfer.c. Аргументы обязательны и не перекрываются. + * request не меняется; response получает копию запроса и результат, status/data_len определяют пригодность данных. + * Одинаковый ID/содержимое мутации возвращает квитанцию; конфликт — BAD_PACKET. После окна гарантий повтора нет. + * Полный кеш: BUSY.cursor — остаток до ближайшего освобождения в timebase; квитанции не вытесняются. */ void mf_storage_execute(struct mf_storage* storage, uint64_t group, uint64_t peer, const struct mf_message* request, struct mf_message* response); /* Диагностика/тест: число файловых дескрипторов, исключая каталог корня. */ diff --git a/src/member_files/transfer.c b/src/member_files/transfer.c index 8a741ab3..63bc8a51 100644 --- a/src/member_files/transfer.c +++ b/src/member_files/transfer.c @@ -36,6 +36,7 @@ struct mf_transfer { EVP_MD_CTX* digest; int fd, server, sending, starting, active, opened, busy, stopping, finished_io, peer_down; int status, work_status, reply_ready, reply_final, verified, committed, end_sent, end_received; + int temp_owned; // Только свой эксклюзивно созданный локальный temp можно публиковать/удалять. enum transfer_work work; uint64_t size, offset, received, read_offset, io_offset, io_bytes, io_size, progress_tb, last_event, last_log, logged_offset; unsigned head, count, io_head, io_count; @@ -59,6 +60,7 @@ static void transfer_event(struct mf_transfer* t, int finished) { } static void transfer_wake(struct mf_transfer* t) { + if (t->stopping && !t->reply_ready) return; if (t->pump || t->waiter.internal || t->waiter.call_soon_id || t->files->closing) return; t->pump = uasync_call_soon(t->files->inst->ua, t, transfer_pump); if (!t->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: stream pump allocation failed id=%llu", (unsigned long long)t->request.id); @@ -129,7 +131,12 @@ static void transfer_work(void* arg) { struct mf_disk_info after; if (!rc) rc = mf_disk_info(t->fd, &after); if (!rc && memcmp(&after, &t->source, sizeof(after))) rc = MF_STALE; - } else rc = mf_disk_local_open(t->temp, 1, &t->fd); + } else { + rc = mf_disk_local_open(t->temp, 1, &t->fd); + if (!rc) t->temp_owned = 1; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: download temp id=%llu created=%d status=%s path=%s", + (unsigned long long)t->request.id, t->temp_owned, mf_status_text(rc), t->temp); + } if (!rc && t->request.upload == 0) { t->digest = EVP_MD_CTX_new(); if (!t->digest || EVP_DigestInit_ex(t->digest, EVP_sha256(), NULL) != 1) rc = MF_IO_ERROR; @@ -196,8 +203,13 @@ static void transfer_work(void* arg) { if (!rc) rc = closed; } if (!t->server && !t->sending) { - if (!t->status && !rc) rc = mf_disk_local_publish(t->temp, t->request.local); - if (t->status || rc) mf_disk_local_remove(t->temp); + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: download cleanup id=%llu owned=%d status=%s io=%s", + (unsigned long long)t->request.id, t->temp_owned, mf_status_text(t->status), mf_status_text(rc)); + if (!t->status && !rc) { + rc = mf_disk_local_publish(t->temp, t->request.local); + if (!rc) t->temp_owned = 0; + } + if ((t->status || rc) && t->temp_owned && !mf_disk_local_remove(t->temp)) t->temp_owned = 0; } EVP_MD_CTX_free(t->digest); t->digest = NULL; } @@ -259,7 +271,9 @@ static void transfer_stop(struct mf_transfer* t, int status) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream stopped id=%llu peer=%llx reason=%s", (unsigned long long)t->request.id, (unsigned long long)t->request.server, mf_status_text(status)); } - t->stopping = 1; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: stream stopping id=%llu starting=%d active=%d io=%d pending=%d", + (unsigned long long)t->request.id, t->starting, t->active, t->busy, t->pending != NULL); + t->stopping = 1; t->starting = 0; if (t->pump) { uasync_call_soon_cancel(t->files->inst->ua, t->pump); t->pump = NULL; } etcp_router_cancel_send_ready(t->files->inst, t->request.group, t->request.server, ETCP_RT_ID_MEMBER_FILES, &t->waiter); if (t->pending) { struct mf_pending* p = t->pending; t->pending = NULL; member_files_cancel(p); } diff --git a/src/member_files/worker.h b/src/member_files/worker.h index 3b9fb0c2..2591b7c5 100644 --- a/src/member_files/worker.h +++ b/src/member_files/worker.h @@ -1,21 +1,32 @@ /* Единственный дисковый поток модуля. submit/done выполняются в uasync, work — * в pthread. Очередь ограничена; completion резервируется до передачи работы. * Для обязательной очистки reserve выполняется до открытия файлов. destroy вне - * callbacks ждёт работ и доставляет done, включая поставленные ими cleanup jobs. */ + * callbacks ждёт работ и доставляет done, включая поставленные ими cleanup jobs. + * data/arg заимствованы до done; worker освобождает только собственную оболочку работы. + * ua/work/done и worker обязательны; ua живёт до конца destroy. Останавливать worker из done нельзя. */ #ifndef MEMBER_FILES_WORKER_H #define MEMBER_FILES_WORKER_H struct UASYNC; struct mf_worker; struct mf_work; typedef void (*mf_work_fn)(void* arg); +/* error=0 — work завершён, результат операции находится в data; -1 — обычная задача не принята (синхронно). */ typedef void (*mf_done_fn)(void* arg, int error); +/* Создать pthread/mutex/condition; возвращает owner либо NULL, без дискового I/O. */ struct mf_worker* mf_worker_create(struct UASYNC* ua); +/* Резерв оболочки и completion, NULL при OOM. Сам по себе не занимает очередь/не задаёт worker. */ struct mf_work* mf_worker_reserve(void); +/* Освободить только ещё не переданный в submit_reserved резерв; NULL допустим. */ void mf_worker_unreserve(struct mf_work* work); +/* До64 обычных задач, включая ожидающие done. При отказе done(-1) синхронно, work не вызывается. */ void mf_worker_submit(struct mf_worker* worker, mf_work_fn work, void* data, mf_done_fn done, void* arg); +/* Передать резерв worker без проверки лимита64: только обязательные cleanup continuations. + * После вызова резерв нельзя освобождать/использовать повторно. Work выполняется FIFO, done ровно один раз. */ void mf_worker_submit_reserved(struct mf_worker* worker, struct mf_work* reserved, mf_work_fn work, void* data, mf_done_fn done, void* arg); // После quiesce владельца, вне callbacks: дождаться работ и их cleanup continuations, не останавливая pthread. +// Не запускает uasync рекурсивно; порядок done при drain не обещается. NULL допустим. void mf_worker_drain(struct mf_worker* worker); +/* Drain, затем join и освобождение. Работы не прерываются и получают обычный done(0); NULL допустим. */ void mf_worker_destroy(struct mf_worker* worker); #endif diff --git a/tests/test_member_files.c b/tests/test_member_files.c index 272aad77..f55b0cfb 100644 --- a/tests/test_member_files.c +++ b/tests/test_member_files.c @@ -14,6 +14,10 @@ #include #include #include +#ifndef _WIN32 +#include +#include +#endif static int failures; #define EXPECT(condition, name) do { if (!(condition)) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "FAIL: %s line=%d", name, __LINE__); failures++; } } while (0) @@ -179,6 +183,28 @@ static void storage_test(const char* directory) { } #ifndef _WIN32 +/* Ошибочный local source не должен блокировать единственный disk worker до проверки типа. */ +static void local_fifo_test(const char* directory) { + char path[1200]; snprintf(path, sizeof(path), "%s/local-fifo", directory); + EXPECT(!mkfifo(path, 0600), "FIFO source fixture"); + pid_t child = fork(); + EXPECT(child >= 0, "FIFO source child"); + if (!child) { + alarm(2); + int fd = -1; struct mf_disk_info info; + int opened = mf_disk_local_open(path, 0, &fd); + int valid = opened == MF_OK && mf_disk_info(fd, &info) == MF_INVALID_PATH; + if (fd >= 0 && mf_disk_close(fd)) valid = 0; + _exit(valid ? 0 : 1); + } + if (child > 0) { + int status = 0; + EXPECT(waitpid(child, &status, 0) == child && WIFEXITED(status) && !WEXITSTATUS(status), "FIFO open returns before type rejection"); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: FIFO source child status=%d; expected normal exit before2s", status); + } + test_unlink(path); +} + static void large_offset_test(const char* directory) { char path[1200]; snprintf(path, sizeof(path), "%s/large-offset", directory); int fd = -1; uint64_t end = (1ULL << 33) + 4; @@ -393,10 +419,11 @@ int main(void) { char directory[1024] = "/tmp/utun_member_files_XXXXXX"; if (test_mkdtemp(directory)) return 1; protocol_test(); storage_test(directory); avatar_storage_test(directory); quota_test(directory); - count_test(directory); quota_scan_failure_test(directory); worker_test(); + count_test(directory); quota_scan_failure_test(directory); #ifndef _WIN32 - large_offset_test(directory); + local_fifo_test(directory); large_offset_test(directory); #endif + worker_test(); printf("member_files: %s failures=%d directory=%s\n", failures ? "FAIL" : "PASS", failures, directory); return failures ? 1 : 0; } diff --git a/tests/test_member_files_network.c b/tests/test_member_files_network.c index 4d7ab760..ec0fdd0e 100644 --- a/tests/test_member_files_network.c +++ b/tests/test_member_files_network.c @@ -51,6 +51,8 @@ struct network_test { unsigned restart_tick; unsigned preservation_case; char preserved_temp[864]; + unsigned mutation_index, busy_responses, dropped_response; + etcp_recv_fn reader_receive; struct api_case api[8]; void* timer; }; @@ -163,6 +165,30 @@ static void response(void* arg, const struct mf_message* r) { struct network_test* t = arg; t->response = *r; t->answered = 1; } +/* Потеря первого результата проверяет повтор с тем же ID без повторного mkdir. */ +static void mutation_receive(struct ETCP_CONN* conn, struct ll_entry* e) { + struct network_test* t = active; + struct mf_message m; + if (e && e->dgram && e->len > ROUTER_SVC_PAYLOAD_OFF && + !mf_decode(e->dgram + ROUTER_SVC_PAYLOAD_OFF, e->len - ROUTER_SVC_PAYLOAD_OFF, &m) && + m.kind == MF_RESPONSE && m.command == MF_MKDIR && m.request_id >= 1000 && m.request_id < 1076) { + if (m.status == MF_BUSY) t->busy_responses++; + if (!m.status && m.request_id == 1000 && !t->dropped_response) { + t->dropped_response = 1; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: dropped successful mutation response id=1000; waiting for replay"); + queue_dgram_free(e); queue_entry_free(e); return; + } + } + t->reader_receive(conn, e); +} + +static void mutation_start(struct network_test* t) { + struct mf_message q = { .command = t->phase == 26 ? MF_STAT : MF_MKDIR, .owner = t->ids[READER], + .request_id = t->phase == 26 ? 2000 + 2 * t->mutation_index : 1000 + t->mutation_index }; + snprintf(q.path, sizeof(q.path), "bulk_%02u", t->mutation_index); t->answered = 0; + if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) fail(t, "bulk mutation start"); +} + static int put_member(struct network_test* t, int destination, int member, int deleted) { uint8_t msg[512], join[64], tags_sig[64], tree[64]; uint64_t ts = 100000; @@ -238,7 +264,7 @@ static int downloaded(struct network_test* t) { static void tick(void* arg) { struct network_test* t = arg; t->timer = NULL; - if (++t->ticks > 6000) { fail(t, "60s deadline"); return; } + if (++t->ticks > 30000) { fail(t, "300s deadline including receipt expiry"); return; } // Отсчитываем полные 6с от фактической блокировки worker, не от постановки задания. pthread_mutex_lock(&t->disk_mutex); if (t->disk_started_tb && !t->disk_released && get_time_tb() - t->disk_started_tb >= 60000) { @@ -251,10 +277,14 @@ static void tick(void* arg) { struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES); if (route->inflight_q->count > t->peak_inflight) t->peak_inflight = route->inflight_q->count; if (t->phase == 0) { - if (t->nodes[SERVER]->member_files->storage_status == MF_OK) { - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: storage ready; upload creates its own CM route request"); - t->phase = 1; transfer(t, 1, 1); + if (t->ticks == 1) { + struct member_files* files = t->nodes[SERVER]->member_files; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: startup tick=%u storage_status=%d closing=%d server=%llx", + t->ticks, files->storage_status, files->closing, (unsigned long long)t->ids[SERVER]); } + int a = etcp_router_get_path(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES).next_hop_node_id != 0; + int b = etcp_router_get_path(t->nodes[READER], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES).next_hop_node_id != 0; + if (a && b && t->nodes[SERVER]->member_files->storage_status == MF_OK) { t->phase = 1; transfer(t, 1, 1); } } else if (t->phase == 1 && t->finished) { if (t->transfer_status) { fail(t, mf_status_text(t->transfer_status)); return; } t->phase = 2; request(t, MF_STAT, t->ids[AUTHOR]); @@ -332,6 +362,39 @@ static void tick(void* arg) { fail(t, "download refusal/cancel removed unowned temp or changed target"); return; } if (++t->preservation_case < 3) { preservation_start(t); goto again; } + t->phase = 24; + t->reader_receive = t->nodes[READER]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES]; + t->nodes[READER]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES] = mutation_receive; + mutation_start(t); + } else if (t->phase == 24 && t->answered) { + if (t->response.status) { fail(t, "bulk mutation did not survive busy/replay"); return; } + if (++t->mutation_index < 76) { mutation_start(t); goto again; } + t->nodes[READER]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES] = t->reader_receive; + if (!t->dropped_response || !t->busy_responses) { fail(t, "bulk mutation did not exercise lost result and busy"); return; } + t->phase = 25; request(t, MF_QUOTA, t->ids[READER]); + } else if (t->phase == 25 && t->answered) { + if (t->response.status || t->response.data_len != MF_QUOTA_BYTES || mf_get64(t->response.data) || + mf_get64(t->response.data + 16) || mf_get64(t->response.data + 24) || mf_get64(t->response.data + 32) != 76) { + fail(t, "bulk mutation duplicated or lost directories"); return; + } + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: PASS76 mutations, lost response replayed once, busy_responses=%u directories=76", + t->busy_responses); + t->phase = 26; t->mutation_index = 0; mutation_start(t); + } else if (t->phase == 26 && t->answered) { + if (t->response.status) { fail(t, "bulk removal did not survive busy"); return; } + if (t->response.command == MF_STAT) { + struct mf_message q = t->response; q.command = MF_REMOVE_PATH; q.request_id++; t->answered = 0; + if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) fail(t, "bulk remove start"); + goto again; + } + if (++t->mutation_index < 76) { mutation_start(t); goto again; } + t->phase = 27; request(t, MF_QUOTA, t->ids[READER]); + } else if (t->phase == 27 && t->answered) { + if (t->response.status || t->response.data_len != MF_QUOTA_BYTES || mf_get64(t->response.data) || + mf_get64(t->response.data + 16) || mf_get64(t->response.data + 24) || mf_get64(t->response.data + 32)) { + fail(t, "bulk removal left directories or quota"); return; + } + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: PASS76 removals, directories=0, used=0, reserved=0"); t->phase = 10; t->answered = 0; struct mf_message q = { .command = MF_STAT, .owner = t->ids[AUTHOR] }; snprintf(q.path, sizeof(q.path), "cancelled.bin"); if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) { fail(t, "cancelled STAT"); return; }