From 7b04a8f4fbeb9c82fc9e696b9893fcc9bf6ffdfb Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 09:42:05 -0500 Subject: [PATCH 1/9] fix: Impose whole-directory size limit; apply 18h limit to pending dir refs: RUM-16880 This PR addresses a few gaps in storage-thread functionality that could lead to runaway disk usage in certain situations, such as when a the tracking consent value never changes from `pending`. ### Problems Compared to the mobile SDKs, the C++ SDK has the following limitations: 1. Age-based file eviction (for files older than 18h) only runs on the granted-consent directory; there's no such mechanism bounding the growth of the pending-consent directory. 2. The C++ SDK imposes no whole-directory disk usage limitation- on mobile platforms, no single feature's event storage directory is allowed to exceed 512 MB. 3. Less critically: whereas the mobile SDKs impose a limit on the maximum size of a single event (no event can exceed 512 kB on Android; 1MB on iOS), the C++ SDK only applies a 4MB limit to the entire batch The combination of issues 1 and 2 means that current versions of the C++ SDK will allow batches of event data stored in the pending-consent directory to remain on disk, causing overall disk usage to grow unbounded. ### Fixes This PR addresses all three of these problems. Key code changes: - `IFilesystem` now supports `GetFileSize()`, implemented via `stat()` or `GetFileAttributesExW()` - In the upload thread, when actually processing batch files for upload, we still reject (and delete) files older than 18h, just as we did before- applying the 18h limit to the consent-granted directory is a natural side effect of the ordinary file age check on upload. - We now _also_ apply the 18h limit to the consent-pending directory, as an explicit check that's run on a per-feature basis, also in the upload thread, every time an upload cycle is invoked for that feature, regardless of whether anything was uploaded. - In the storage thread, when we're preparing a brand new batch file for write, we now stat all existing batch files in the directory (ignoring files with non-numeric names entirely), check if they sum to greater than 512 MB, and iteratively delete files (oldest-first) until we're back under that limit. - In the storage thread, when we write any event, we now enforce a maximum per-event size limit of 512 kB. As a result, when the SDK is operated normally with the same set of features registered, it will maintain an overall disk usage footprint no greater than `512 MB * N`, where N is the number of features registered. Refactoring changes: - In `upload_thread.cpp`, we now check file age in two separate code paths (normal upload and explicit pending-dir eviction check), so we use a new `batch_file_age()` helper to compute the age of a file given its filename and the current time - In `storage_write.cpp`, when we need to cut a new batch file, we preemptively list the names of all files in the directory we're about to write to. Previously, this list of filenames was just used to prevent conflicts when naming the new file. As of this PR, we _also_ need to use this list of filenames when performing the size-based eviction check. As a result, our directory listing (via `CacheKnownFilenames()`) has been pulled up to a higher scope, and we rely on the fact that `_last_known_filenames` is pre-populated in both of those lower-level routines: - `CacheKnownFilenames()` runs first, populating the set of filenames, filtering it to only numerically-named files, and sorting it - `PurgeDirectoryIfNeeded()` runs next, stat'ing all files in that list; if it deletes old files; it removes them from the front of the list, mutating `_last_known_filenames` - `GetFilenameForNextWrite()` runs next, using `_last_known_filenames` as-is (so we only fetch the directory listing once per new batch file creation) Test changes: - `MockFilesystem` now supports `GetFileSize()`, including targeted mock failures - `storage_write_test.cpp` now validates: - Single event exceeding max event size is dropped - Single event exactly matching max event size is accepted - If batch dir exceeds directory size limit on write, old files are purged - If batch dir is within size limit on write, no files are purged - If foreign (non-SDK-written) files exist in batch dir, they are ignored on purge - Both pending and granted directories are subject to file size limit and will be purged the same - Filesystem errors in size checks are handled gracefully - `upload_thread_test.cpp` now validates: - If a pending dir has files older than 18h when an upload cycle runs, those files are deleted - If all files in a pending dir are newer than 18h, the pendir dir is left untouched - Foreign files older than 18h are left alone, never deleted - If we fail to list files at the start of an eviction check, we skip the age-based eviction process entirely --- src/datadog/impl/core/storage/filesystem.hpp | 13 + .../impl/core/storage/filesystem_posix.cpp | 8 + .../impl/core/storage/filesystem_windows.cpp | 11 + .../impl/core/storage/filesystem_wrapper.cpp | 7 + .../impl/core/storage/filesystem_wrapper.hpp | 1 + src/datadog/impl/core/storage_write.cpp | 119 ++++++++-- src/datadog/impl/core/storage_write.hpp | 50 +++- src/datadog/impl/core/upload_thread.cpp | 90 +++++-- tests/impl/core/storage_write_test.cpp | 223 ++++++++++++++++-- tests/impl/core/upload_thread_test.cpp | 196 +++++++++++++++ tests/mock/filesystem.cpp | 16 +- tests/mock/filesystem.hpp | 2 + 12 files changed, 664 insertions(+), 72 deletions(-) diff --git a/src/datadog/impl/core/storage/filesystem.hpp b/src/datadog/impl/core/storage/filesystem.hpp index 47c17f41..f67cb3f8 100644 --- a/src/datadog/impl/core/storage/filesystem.hpp +++ b/src/datadog/impl/core/storage/filesystem.hpp @@ -206,6 +206,19 @@ class IFilesystem { */ virtual FilesystemResult Close(PlatformFileHandle handle) = 0; + struct GetFileSizeResult { + FilesystemResult value; + size_t size; // valid only when value == OK + }; + + /** + * Returns the size of the regular file at `path` in bytes. + * + * On success, returns {OK, size}. If no file exists at the path, returns + * {DoesNotExist, 0}. Any other error code represents a failure. + */ + virtual GetFileSizeResult GetFileSize(const PlatformPath& path) = 0; + /** * Deletes a regular file. * diff --git a/src/datadog/impl/core/storage/filesystem_posix.cpp b/src/datadog/impl/core/storage/filesystem_posix.cpp index 318431e3..57dc95fc 100644 --- a/src/datadog/impl/core/storage/filesystem_posix.cpp +++ b/src/datadog/impl/core/storage/filesystem_posix.cpp @@ -306,6 +306,14 @@ class PosixFilesystem final : public IFilesystem { return map_errno(errno); } + GetFileSizeResult GetFileSize(const PlatformPath& path) override { + struct stat st; + if (stat(path.Get(), &st) != 0) { + return {map_errno(errno), 0}; + } + return {FilesystemResult::OK, static_cast(st.st_size)}; + } + FilesystemResult Delete(const PlatformPath& path) override { const int result = unlink(path.Get()); if (result == 0) { diff --git a/src/datadog/impl/core/storage/filesystem_windows.cpp b/src/datadog/impl/core/storage/filesystem_windows.cpp index b9c71dbc..23dc2f64 100644 --- a/src/datadog/impl/core/storage/filesystem_windows.cpp +++ b/src/datadog/impl/core/storage/filesystem_windows.cpp @@ -352,6 +352,17 @@ class WindowsFilesystem final : public IFilesystem { return FilesystemResult::OK; } + GetFileSizeResult GetFileSize(const PlatformPath& path) override { + WIN32_FILE_ATTRIBUTE_DATA data; + if (GetFileAttributesExW(path.Get(), GetFileExInfoStandard, &data) == 0) { + return {map_error(GetLastError()), 0}; + } + const size_t size = static_cast( + (static_cast(data.nFileSizeHigh) << 32) | data.nFileSizeLow + ); + return {FilesystemResult::OK, size}; + } + FilesystemResult Delete(const PlatformPath& path) override { const BOOL result = DeleteFileW(path.Get()); if (result == 0) { diff --git a/src/datadog/impl/core/storage/filesystem_wrapper.cpp b/src/datadog/impl/core/storage/filesystem_wrapper.cpp index 7da6a89a..0bb593ed 100644 --- a/src/datadog/impl/core/storage/filesystem_wrapper.cpp +++ b/src/datadog/impl/core/storage/filesystem_wrapper.cpp @@ -116,6 +116,13 @@ FilesystemWrapper::OpenFileResult FilesystemWrapper::OpenForRead( return {res.value, File{_fs, res.handle}}; } +IFilesystem::GetFileSizeResult FilesystemWrapper::GetFileSize(const char* path) { + if (!_path.Encode(path)) { + return {FilesystemResult::PathEncodingFailed, 0}; + } + return _fs.GetFileSize(_path); +} + FilesystemResult FilesystemWrapper::Delete(const char* path) { if (!_path.Encode(path)) { return FilesystemResult::PathEncodingFailed; diff --git a/src/datadog/impl/core/storage/filesystem_wrapper.hpp b/src/datadog/impl/core/storage/filesystem_wrapper.hpp index 31852cbf..131e0305 100644 --- a/src/datadog/impl/core/storage/filesystem_wrapper.hpp +++ b/src/datadog/impl/core/storage/filesystem_wrapper.hpp @@ -96,6 +96,7 @@ class FilesystemWrapper { ); OpenFileResult OpenForWrite(const char* path, bool append, bool hold_advisory_lock); OpenFileResult OpenForRead(const char* path, bool hold_advisory_lock); + IFilesystem::GetFileSizeResult GetFileSize(const char* path); FilesystemResult Delete(const char* path); FilesystemResult DeleteDirectory(const char* path); FilesystemResult Rename(const char* src, const char* dst); diff --git a/src/datadog/impl/core/storage_write.cpp b/src/datadog/impl/core/storage_write.cpp index b90ca7f6..5fb51f3d 100644 --- a/src/datadog/impl/core/storage_write.cpp +++ b/src/datadog/impl/core/storage_write.cpp @@ -8,6 +8,7 @@ #include #include +#include #include #include #include @@ -162,12 +163,12 @@ bool BatchWriter::FlushEvent( num_bytes += TLVBlockHeader::SIZE + event_metadata.size(); } - // If the encoded size of this write (metadata + event, with headers) exceeds the - // maximum configured size for a single batch file, reject the event outright: even a - // brand new batch file could not contain it - if (num_bytes > _config.max_file_size) { + // If the encoded size of this single event (metadata + event, with headers) exceeds + // the per-event limit, reject it outright + if (num_bytes > _config.max_event_size) { _diagnostic_logger.Error( - "Event dropped; size of single event exceeds max batch file size" + "Event dropped; size of single event exceeds max event size", + {{"event_size", num_bytes}, {"max_event_size", _config.max_event_size}} ); return false; } @@ -288,11 +289,21 @@ BatchWriter::FileDetails* BatchWriter::PrepareFileForNextWrite( return &active_file; } - // If not, prepare to write a new file: start by figuring out what to name it - const auto next = GetFilenameForNextWrite(consent_dir_path, current_time); + // If we can't reuse the last file, prepare to write a new file. List the directory + // contents once here so that both the quota check and the filename search can share + // the same listing. + if (!CacheKnownFilenames(consent_dir_path)) { + active_file.Clear(); + return nullptr; + } + + // Enforce the directory size quota, using and updating the cached listing + PurgeDirectoryIfNeeded(consent_dir_path); + + // Figure out what to name the new file, using the post-purge cached listing + const auto next = GetFilenameForNextWrite(current_time); if (!next) { - // Failed to resolve new filename (i.e. listing directory contents failed, or all - // potential filenames are in use); can't proceed with file creation + // All potential filenames are in use; can't proceed with file creation active_file.Clear(); return nullptr; } @@ -354,16 +365,11 @@ bool BatchWriter::CanReuseFileForNextWrite( } std::optional> BatchWriter::GetFilenameForNextWrite( - const StoragePath& consent_dir_path, Timestamp current_time + Timestamp current_time ) const { - // Our new file will use a filename that reflects the current timestamp, but it's - // possible that a file already exists with that name, and we don't want to reuse any - // existing files that we didn't create ourselves: so start by retrieving the list of - // existing filenames in our target directory - if (!CacheKnownFilenames(consent_dir_path)) { - // Failed to list directory contents; can't proceed with file creation - return std::nullopt; - } + // Our new file will use a filename reflecting the current timestamp. If a file with + // that name already exists (per the pre-populated _last_known_filenames), we + // increment by 1ms at a time to find a free slot, up to a limit of 100 attempts. // Use stack memory to convert uint64 to string for candidate filenames std::array buffer{0}; @@ -427,9 +433,84 @@ bool BatchWriter::CacheKnownFilenames(const StoragePath& consent_dir_path) const return false; } - // Sort filenames for deterministic iteration in timestamp-name-order + // Sort filenames for deterministic iteration in timestamp-name-order, then strip any + // non-numeric names (files not written by the SDK) so that both PurgeDirectoryIfNeeded + // and GetFilenameForNextWrite operate only on batch files we own. std::sort(_last_known_filenames.begin(), _last_known_filenames.end()); + _last_known_filenames.erase( + std::remove_if( + _last_known_filenames.begin(), + _last_known_filenames.end(), + [](const std::string& name) { + return !std::all_of(name.begin(), name.end(), [](unsigned char c) { + return std::isdigit(c); + }); + } + ), + _last_known_filenames.end() + ); return true; } +void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) { + // _last_known_filenames is pre-populated, sorted, and already filtered to numeric-named + // (SDK-written) batch files by CacheKnownFilenames. + + // Accumulate sizes for each file using a reusable path buffer. Files whose size cannot + // be determined are treated as zero: they don't push the running total toward the quota + // threshold, but remain eligible for deletion if the purge loop runs and reaches them + // (without contributing to total reduction when deleted). + FilesystemWrapper fsw(_fs); + StoragePath path_buf; + std::vector sizes; + sizes.reserve(_last_known_filenames.size()); + size_t total = 0; + for (const std::string& name : _last_known_filenames) { + path_buf.MustSet(dir); + if (!path_buf.Append(name)) { + sizes.push_back(0); + continue; + } + const auto res = fsw.GetFileSize(path_buf.CStr()); + const size_t file_size = res.value == FilesystemResult::OK ? res.size : 0; + sizes.push_back(file_size); + total += file_size; + } + + if (total <= _config.max_directory_size) { + return; + } + + // Delete oldest files until we're within quota, erasing each from _last_known_filenames + // and the parallel sizes vector in lock-step so GetFilenameForNextWrite sees the + // post-purge state. Iterate by index rather than iterator so we can erase in-place: + // on a successful deletion we don't advance i (the next element slides into position); + // on failure we skip the file and increment. + size_t i = 0; + while (i < _last_known_filenames.size() && total > _config.max_directory_size) { + path_buf.MustSet(dir); + if (!path_buf.Append(_last_known_filenames[i])) { + ++i; + continue; + } + const auto delete_res = fsw.Delete(path_buf.CStr()); + if (delete_res == FilesystemResult::OK) { + total -= sizes[i]; + _diagnostic_logger.Debug( + "Deleted batch file to enforce directory size quota", + {{"path", path_buf.Get()}} + ); + const auto idx = static_cast(i); + _last_known_filenames.erase(_last_known_filenames.begin() + idx); + sizes.erase(sizes.begin() + idx); + } else { + _diagnostic_logger.Warning( + "Failed to delete batch file during directory quota enforcement", + {{"path", path_buf.Get()}, {"error", FilesystemResultStr(delete_res)}} + ); + ++i; + } + } +} + } // namespace datadog::impl diff --git a/src/datadog/impl/core/storage_write.hpp b/src/datadog/impl/core/storage_write.hpp index cf5d346b..cfceaab9 100644 --- a/src/datadog/impl/core/storage_write.hpp +++ b/src/datadog/impl/core/storage_write.hpp @@ -37,6 +37,18 @@ struct BatchWriterConfig { * bytes), we will not write to that file and will instead create a new one. */ size_t max_file_size{0x400000}; // 4 MB + /** + * If the encoded size of a single event write (metadata + event, with TLV headers) + * exceeds this threshold, the event is dropped rather than written. + */ + size_t max_event_size{512UL * 1024}; // 512 KB + /** + * Maximum total size (in bytes) of all batch files in a single consent-level + * directory (pending or granted) for one feature. When a new batch file is about to + * be created and the directory already exceeds this limit, the oldest files are + * deleted until the total is within the quota. + */ + size_t max_directory_size{512ULL * 1024 * 1024}; // 512 MB /** * Maximum number of events that we will write to a single file. A single write * operation may include both metadata and event blocks: i.e. if configured with a @@ -115,8 +127,9 @@ class BatchWriter { /** * Buffers the set of filenames retrieved in our last call to CacheKnownFilenames(). - * Contains the names of all files present in the target directory for the current - * consent value. Reset when tracking consent changes. + * Contains the names of all SDK-written (numeric-named) batch files present in the + * target directory for the current consent value, sorted in ascending order. Reset + * when tracking consent changes. */ mutable std::vector _last_known_filenames; @@ -240,19 +253,40 @@ class BatchWriter { ) const; /** - * Designates a name to use for a newly-created batch file in the given directory, - * returning both the integer millisecond count reflecting file creation time, as well - * as the string-formatted version of that same value. + * Designates a name for a newly-created batch file, returning both the integer + * millisecond count reflecting file creation time and the string-formatted value. + * + * Requires that `_last_known_filenames` has already been populated by + * `CacheKnownFilenames` before this function is called. */ std::optional> GetFilenameForNextWrite( - const StoragePath& consent_dir_path, Timestamp current_time + Timestamp current_time ) const; /** - * Retrieves a list of all files in the current directory, caching that set of names - * in _last_known_filenames. + * Populates `_last_known_filenames` with a sorted listing of all numeric-named + * (SDK-written) regular files in `consent_dir_path`, excluding any foreign files + * whose names contain non-digit characters. Must be called in + * `PrepareFileForNextWrite` before invoking `PurgeDirectoryIfNeeded` or + * `GetFilenameForNextWrite`. + * + * Returns true on success, false if the directory listing fails. */ bool CacheKnownFilenames(const StoragePath& consent_dir_path) const; + + /** + * Enforces the per-directory size quota for `dir` by deleting the oldest batch files + * until the total size of remaining files falls within `_config.max_directory_size`. + * + * Reads and mutates `_last_known_filenames` (pre-populated and pre-filtered to + * numeric-named files by `CacheKnownFilenames`), removing deleted entries so that + * `GetFilenameForNextWrite` sees an accurate post-purge listing. + * + * Called only when a new batch file is about to be created, mirroring the iOS SDK's + * approach of deferring the quota check to file-creation time. The check is + * best-effort: errors from sizing individual files are tolerated. + */ + void PurgeDirectoryIfNeeded(const StoragePath& dir); }; } // namespace datadog::impl diff --git a/src/datadog/impl/core/upload_thread.cpp b/src/datadog/impl/core/upload_thread.cpp index 1b6cce1d..889571b6 100644 --- a/src/datadog/impl/core/upload_thread.cpp +++ b/src/datadog/impl/core/upload_thread.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -165,6 +166,23 @@ static _process_and_upload_batch_result _process_and_upload_batch( return _interpret_http_result(res); } +// Parses a batch filename as a millisecond-precision Unix timestamp and returns its age +// relative to `now`. Returns nullopt for non-numeric filenames (non-SDK files that +// should be left alone). Guards against underflow if the timestamp is in the future by +// clamping the result to zero. +static std::optional batch_file_age(const std::string& filename, Timestamp now) { + uint64_t timestamp_ms{0}; + const auto parse_result = std::from_chars( + filename.data(), filename.data() + filename.size(), timestamp_ms + ); + if (parse_result.ec != std::errc{} || + parse_result.ptr != filename.data() + filename.size()) { + return std::nullopt; + } + const Timestamp file_time{std::chrono::milliseconds(timestamp_ms)}; + return file_time < now ? now - file_time : Duration::zero(); +} + static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-complexity) DiagnosticLogger& diagnostic_logger, UploadThreadConfig config, @@ -211,40 +229,19 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp _process_and_upload_batch_result last_batch_result{ _process_and_upload_batch_result::success }; + const Timestamp now = clock.Now(); for (const std::string& filename : mut_filenames) { - // If a filename is all-integer, it's taken to be a Unix timestamp in milliseconds, - // corresponding to the system_clock value that the storage thread read when it - // created the file - uint64_t timestamp_ms{0}; - const auto parse_result = std::from_chars( - filename.data(), filename.data() + filename.size(), timestamp_ms - ); - - // Skip any files whose names are not strictly numeric - const bool parse_ok = parse_result.ec == std::errc{}; - const bool parsed_full_string = - parse_result.ptr == (filename.data() + filename.size()); - if (!parse_ok || !parsed_full_string) { + const auto file_age = batch_file_age(filename, now); + if (!file_age) { continue; } - // Read the system clock so we can determine the relative age of the file - const Timestamp file_time{std::chrono::milliseconds(timestamp_ms)}; - const Timestamp now = clock.Now(); - - // Defensive: guard against integer underflow on file age calculation. If the file - // appears to be from the future, clamp its age to 0. - Duration file_age = Duration::zero(); - if (file_time < now) { - file_age = now - file_time; - } - // Under normal circumstances, we have to respect a minimum file age threshold // before we can read from files, as the storage thread may still be writing to them // before they hit that age. If our threshold is 0, it means we're permitted to // bypass these checks and read from all files, as long as they're not too _old_ to // upload. - if (file_age < config.min_file_age_for_read) { + if (*file_age < config.min_file_age_for_read) { // If we've encountered our first file that's too new to process, we're done. We // process files in order, sorted lexically by timestamp, so every file that we'd // encounter hereafter would be even newer than this one. @@ -262,7 +259,7 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp } // If this file is too old to process, delete it and continue - if (file_age >= config.max_file_age_for_read) { + if (*file_age >= config.max_file_age_for_read) { const auto delete_res = fsw.Delete(file_path.CStr()); if (delete_res == FilesystemResult::OK) { diagnostic_logger.Debug( @@ -364,6 +361,47 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp } } + // Evict stale files from the pending-consent directory. These accumulate when consent + // is set to Pending and never resolved. The upload thread never reads from this + // directory, so its only clean-up mechanism is this age-based eviction pass. + const StoragePath& pending_dir_path = feature.storage->GetPendingPath(); + mut_filenames.clear(); + const auto pending_list_res = fsw.ListFiles(pending_dir_path.CStr(), mut_filenames); + if (pending_list_res == FilesystemResult::OK) { + std::sort(mut_filenames.begin(), mut_filenames.end()); + for (const std::string& pending_filename : mut_filenames) { + const auto file_age = batch_file_age(pending_filename, now); + if (!file_age) { + continue; + } + if (*file_age < config.max_file_age_for_read) { + // Files are sorted oldest-first; all subsequent files are also too young. + break; + } + + StoragePath pending_file_path; + pending_file_path.MustSet(pending_dir_path); + if (!pending_file_path.Append(pending_filename)) { + continue; + } + + const auto delete_res = fsw.Delete(pending_file_path.CStr()); + if (delete_res == FilesystemResult::OK) { + diagnostic_logger.Debug( + "Deleted outdated pending batch file", + {{"feature", feature.name}, {"filename", pending_filename}} + ); + } else { + diagnostic_logger.Warning( + "Failed to delete outdated pending batch file", + {{"feature", feature.name}, + {"filename", pending_filename}, + {"error", FilesystemResultStr(delete_res)}} + ); + } + } + } + if (num_uploads_attempted == 0) { diagnostic_logger.Debug( "Upload cycle finished with nothing to upload", {{"feature", feature.name}} diff --git a/tests/impl/core/storage_write_test.cpp b/tests/impl/core/storage_write_test.cpp index 7eaf9096..1d895b0d 100644 --- a/tests/impl/core/storage_write_test.cpp +++ b/tests/impl/core/storage_write_test.cpp @@ -336,24 +336,6 @@ TEST_CASE("BatchWriter", "[unit]") { ); } - SECTION("M drop event W single write exceeds max file size") { - // Given a batch writer configured with a max batch file size of 16 bytes - auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); - config.max_file_size = 16; - BatchWriter writer = init_writer(config); - - // When we attempt to write a 20-byte payload, which is larger than any file - // could contain - const bool write_ok = writer.HandleWrite("twenty-bytes-0", {}, false); - - // Then the event is not accepted - REQUIRE(write_ok == false); - - // And nothing is written to disk - std::vector names = fs.Ls(pending_dir_path); - REQUIRE(names.empty()); - } - SECTION("M start new file W max writes per file exceeded") { // Given a batch writer configured to write no more than 2 events per file auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); @@ -787,6 +769,211 @@ TEST_CASE("BatchWriter", "[unit]") { REQUIRE(names.size() == 1); } + SECTION("M drop event W single write exceeds max_event_size but not max_file_size") { + // Given a batch writer with a generous max_file_size but a tight max_event_size + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_file_size = 1024 * 1024; // 1 MB batch file — plenty of room + config.max_event_size = 20; // only 20 bytes per event + BatchWriter writer = init_writer(config); + + // When we attempt to write a payload where header (6 bytes) + data (15 bytes) = 21 + REQUIRE_FALSE(writer.HandleWrite("fifteen-bytes!!", {}, false)); + + // Then nothing is written to disk + REQUIRE(fs.Ls(pending_dir_path).empty()); + } + + SECTION("M accept event W single write exactly equals max_event_size") { + // Given a batch writer with max_event_size set to exactly fit a 14-byte write + // (6-byte TLV header + 8 bytes data) + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_event_size = 14; + config.max_file_size = 1024 * 1024; + BatchWriter writer = init_writer(config); + + // When we write an 8-byte event (total encoded size = 14 bytes) + REQUIRE(writer.HandleWrite("8-bytes!", {}, false)); + REQUIRE(fs.Ls(pending_dir_path).size() == 1); + } + + SECTION("M purge oldest files W directory size exceeds max_directory_size") { + // Given a batch writer with a directory quota of 50 bytes + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 50; + BatchWriter writer = init_writer(config); + + // And three existing batch files each 40 bytes at timestamps 0, 1, and 2 + // (total 120 bytes, well over the 50-byte quota) + fs.Touch(pending_prefix + "1700000000000", std::string(40, 'x')); + fs.Touch(pending_prefix + "1700000000001", std::string(40, 'y')); + fs.Touch(pending_prefix + "1700000000002", std::string(40, 'z')); + + // When we write a new event 100ms later (triggers new file + quota check) + clock.TickMilliseconds(100); + REQUIRE(writer.HandleWrite("event-new", {}, false)); + + // Then the two oldest files are purged (oldest: 40 bytes → 80 > 50; next: 40 + // bytes → 40 ≤ 50, stop), leaving only the youngest pre-existing file and the + // new batch file + std::vector names = fs.Ls(pending_dir_path); + std::sort(names.begin(), names.end()); + REQUIRE(names.size() == 2); + REQUIRE(names[0] == "1700000000002"); + REQUIRE(names[1] == "1700000000100"); + } + + SECTION("M not purge any files W directory size is within max_directory_size") { + // Given a batch writer with a generous quota + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 10000; + BatchWriter writer = init_writer(config); + + // And a small existing file in the pending directory + fs.Touch(pending_prefix + "1700000000000", std::string(10, 'x')); + + // When we write a new event after triggering a file rotation + clock.TickMilliseconds(100); + REQUIRE(writer.HandleWrite("event-new", {}, false)); + + // Then no files are deleted + std::vector names = fs.Ls(pending_dir_path); + std::sort(names.begin(), names.end()); + REQUIRE(names.size() == 2); + REQUIRE(names[0] == "1700000000000"); + REQUIRE(names[1] == "1700000000100"); + } + + SECTION("M not purge non-numeric-named files W directory contains foreign files") { + // Given a batch writer with a very small quota + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 1; + BatchWriter writer = init_writer(config); + + // And a non-numerically-named file (not written by the SDK) that would push the + // directory over quota if counted + fs.Touch(pending_prefix + "not-a-batch", std::string(100, 'x')); + + // When a new event triggers the quota check + clock.TickMilliseconds(100); + REQUIRE(writer.HandleWrite("event-new", {}, false)); + + // Then the non-SDK file is left untouched + REQUIRE(fs.IsFile(pending_prefix + "not-a-batch")); + } + + SECTION( + "M leave existing files intact W directory listing fails on new file creation" + ) { + // Given a batch writer with a directory quota so tight that any existing file + // would trigger a purge + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 1; + BatchWriter writer = init_writer(config); + + // And an existing batch file in the pending directory + fs.Touch(pending_prefix + "1700000000000", std::string(100, 'x')); + + // And a filesystem that refuses to list the pending directory + fs.SimulateFailure( + pending_dir_path, + FilesystemResult::PermissionDenied, + MockFilesystem::FailureFlags::Ls + ); + + // When we attempt to write an event (no active file, so new-file path is taken and + // the quota check fires). The write also fails because filename resolution requires + // listing the same directory. + clock.TickMilliseconds(100); + writer.HandleWrite("event-new", {}, false); + + // Then the pre-existing file is untouched — purge bailed out rather than + // deleting files based on an incomplete view of the directory + fs.ClearSimulatedFailure(pending_dir_path); + REQUIRE(fs.IsFile(pending_prefix + "1700000000000")); + } + + SECTION( + "M drop event W metadata and event together exceed max_event_size but neither " + "alone would" + ) { + // Given a batch writer where max_event_size = 24 bytes. + // A 5-byte event alone encodes to 11 bytes (6-byte TLV header + 5 data), which is + // within the limit. An 8-byte metadata block encodes to 14 bytes. Combined they + // require 25 bytes, which exceeds the 24-byte cap. + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_file_size = 1024 * 1024; + config.max_event_size = 24; + BatchWriter writer = init_writer(config); + + // When we write the 5-byte event with no metadata, it fits (11 bytes ≤ 24) + REQUIRE(writer.HandleWrite("event", {}, false)); + + // When we write the same event with an 8-byte metadata block, the combined size + // of 25 bytes exceeds the per-event cap + REQUIRE_FALSE(writer.HandleWrite("event", "metadata", false)); + + // Then only the first write produced a file; the second was dropped + REQUIRE(fs.Ls(pending_dir_path).size() == 1); + } + + SECTION("M purge oldest files W quota exceeded in granted directory") { + // Given a batch writer writing to the granted directory with a 50-byte quota + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 50; + BatchWriter writer = init_writer(config, TrackingConsent::Granted); + + // And three existing files of 40 bytes each (total 120 bytes, over the 50-byte + // quota) + fs.Touch(granted_prefix + "1700000000000", std::string(40, 'x')); + fs.Touch(granted_prefix + "1700000000001", std::string(40, 'y')); + fs.Touch(granted_prefix + "1700000000002", std::string(40, 'z')); + + // When we write a new event 100ms later (triggers new file + quota check) + clock.TickMilliseconds(100); + REQUIRE(writer.HandleWrite("event-new", {}, false)); + + // Then the two oldest files are purged, leaving only the youngest pre-existing + // file and the new batch file + std::vector names = fs.Ls(granted_dir_path); + std::sort(names.begin(), names.end()); + REQUIRE(names.size() == 2); + REQUIRE(names[0] == "1700000000002"); + REQUIRE(names[1] == "1700000000100"); + } + + SECTION( + "M treat GetFileSize failure as zero W file stat fails during quota check" + ) { + // Given a batch writer with a 50-byte quota + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 50; + BatchWriter writer = init_writer(config); + + // And an oldest file that is 40 bytes but has a simulated stat failure, so it + // contributes 0 bytes to the quota total + fs.Touch(pending_prefix + "1700000000000", std::string(40, 'x')); + fs.SimulateFailure( + pending_prefix + "1700000000000", + FilesystemResult::UnknownError, + MockFilesystem::FailureFlags::Stat + ); + + // And two younger files of 10 bytes each (true total 60 bytes, but apparent total + // with the stat failure counted as 0 is only 20 bytes — within the 50-byte quota) + fs.Touch(pending_prefix + "1700000000001", std::string(10, 'y')); + fs.Touch(pending_prefix + "1700000000002", std::string(10, 'z')); + + // When we write a new event 100ms later (triggers new file + quota check) + clock.TickMilliseconds(100); + REQUIRE(writer.HandleWrite("event-new", {}, false)); + + // Then no files are purged: the failed stat caused the directory to appear to be + // within quota, so the purge was a no-op + REQUIRE(fs.IsFile(pending_prefix + "1700000000000")); + REQUIRE(fs.IsFile(pending_prefix + "1700000000001")); + REQUIRE(fs.IsFile(pending_prefix + "1700000000002")); + } + SECTION("M write bypass event to granted dir W active file is warm in pending dir") { // Given a batch writer with consent Pending and a prior normal write that has // warmed up _active_file to a file in the pending directory diff --git a/tests/impl/core/upload_thread_test.cpp b/tests/impl/core/upload_thread_test.cpp index 0b46130e..8cff60ab 100644 --- a/tests/impl/core/upload_thread_test.cpp +++ b/tests/impl/core/upload_thread_test.cpp @@ -832,6 +832,202 @@ TEST_CASE("HandleUploadProc", "[unit]") { REQUIRE(delay_until_next_cycle == std::chrono::milliseconds(55000)); } + SECTION("M delete outdated files W pending directory contains stale batch files") { + // Given an upload cycle configuration with an 18-hour TTL + clock.FreezeAtMilliseconds(1700000000000); + MockHttpClient client; + const auto [core_config, config] = init_config( + BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium + ); + auto features = register_feature(UploadFrequency::Average); + + // And a pending directory containing one file that is over 18 hours old and one + // that is fresh + const std::string alpha_pending_dir = + "app/.datadog/main/12345/alpha/intermediate-v1"; + const std::string alpha_pending_prefix = alpha_pending_dir + "/"; + const uint64_t eighteen_hours_ms = 18ULL * 60 * 60 * 1000; + const uint64_t old_ts = 1700000000000ULL - eighteen_hours_ms - 1000; + const uint64_t fresh_ts = 1700000000000ULL - 1000; + fs.Touch( + alpha_pending_prefix + std::to_string(old_ts), + MockTLVFile().AppendEvent("old-event").ToString() + ); + fs.Touch( + alpha_pending_prefix + std::to_string(fresh_ts), + MockTLVFile().AppendEvent("fresh-event").ToString() + ); + + // When we run an upload cycle + std::vector filenames; + std::vector read_buffer; + Internal_HandleUploadProc( + logger, + config, + clock, + CreateFeatureId("ALFA"), + features, + fs, + client, + filenames, + read_buffer + ); + + // Then the stale pending file is deleted + REQUIRE_FALSE(fs.IsFile(alpha_pending_prefix + std::to_string(old_ts))); + + // And the fresh pending file is left intact + REQUIRE(fs.IsFile(alpha_pending_prefix + std::to_string(fresh_ts))); + + // And no upload requests were made (pending files are never uploaded) + REQUIRE(client.requests.empty()); + } + + SECTION( + "M leave pending files intact W they are younger than max_file_age_for_read" + ) { + // Given an upload cycle configuration with an 18-hour TTL + clock.FreezeAtMilliseconds(1700000000000); + MockHttpClient client; + const auto [core_config, config] = init_config( + BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium + ); + auto features = register_feature(UploadFrequency::Average); + + // And a pending directory with two files both younger than 18 hours + const std::string alpha_pending_dir = + "app/.datadog/main/12345/alpha/intermediate-v1"; + const std::string alpha_pending_prefix = alpha_pending_dir + "/"; + fs.Touch( + alpha_pending_prefix + "1699999000000", + MockTLVFile().AppendEvent("event-a").ToString() + ); + fs.Touch( + alpha_pending_prefix + "1699999500000", + MockTLVFile().AppendEvent("event-b").ToString() + ); + + // When we run an upload cycle + std::vector filenames; + std::vector read_buffer; + Internal_HandleUploadProc( + logger, + config, + clock, + CreateFeatureId("ALFA"), + features, + fs, + client, + filenames, + read_buffer + ); + + // Then both pending files remain untouched + REQUIRE(fs.Ls(alpha_pending_dir).size() == 2); + REQUIRE(client.requests.empty()); + } + + SECTION("M not delete non-numeric files W pending directory contains non-SDK files") { + // Given an upload cycle with an 18-hour TTL + clock.FreezeAtMilliseconds(1700000000000); + MockHttpClient client; + const auto [core_config, config] = init_config( + BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium + ); + auto features = register_feature(UploadFrequency::Average); + + const std::string alpha_pending_dir = + "app/.datadog/main/12345/alpha/intermediate-v1"; + const std::string alpha_pending_prefix = alpha_pending_dir + "/"; + + // And a pending directory containing one stale SDK batch file and one + // non-numerically-named file that was not written by the SDK + const uint64_t eighteen_hours_ms = 18ULL * 60 * 60 * 1000; + const uint64_t old_ts = 1700000000000ULL - eighteen_hours_ms - 1000; + fs.Touch( + alpha_pending_prefix + std::to_string(old_ts), + MockTLVFile().AppendEvent("old-event").ToString() + ); + fs.Touch(alpha_pending_prefix + "not-an-sdk-file", "foreign-data"); + + // When we run an upload cycle + std::vector filenames; + std::vector read_buffer; + Internal_HandleUploadProc( + logger, + config, + clock, + CreateFeatureId("ALFA"), + features, + fs, + client, + filenames, + read_buffer + ); + + // Then the stale SDK batch file is evicted + REQUIRE_FALSE(fs.IsFile(alpha_pending_prefix + std::to_string(old_ts))); + + // But the non-SDK file is left untouched + REQUIRE(fs.IsFile(alpha_pending_prefix + "not-an-sdk-file")); + + // And no upload requests were made + REQUIRE(client.requests.empty()); + } + + SECTION( + "M silently skip pending eviction W ListFiles fails on pending directory" + ) { + // Given an upload cycle with an 18-hour TTL + clock.FreezeAtMilliseconds(1700000000000); + MockHttpClient client; + const auto [core_config, config] = init_config( + BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium + ); + auto features = register_feature(UploadFrequency::Average); + + const std::string alpha_pending_dir = + "app/.datadog/main/12345/alpha/intermediate-v1"; + const std::string alpha_pending_prefix = alpha_pending_dir + "/"; + + // And a stale pending file that would normally be evicted + const uint64_t eighteen_hours_ms = 18ULL * 60 * 60 * 1000; + const uint64_t old_ts = 1700000000000ULL - eighteen_hours_ms - 1000; + fs.Touch( + alpha_pending_prefix + std::to_string(old_ts), + MockTLVFile().AppendEvent("old-event").ToString() + ); + + // But the pending directory refuses to be listed + fs.SimulateFailure( + alpha_pending_dir, + FilesystemResult::PermissionDenied, + MockFilesystem::FailureFlags::Ls + ); + + // When we run an upload cycle + std::vector filenames; + std::vector read_buffer; + Internal_HandleUploadProc( + logger, + config, + clock, + CreateFeatureId("ALFA"), + features, + fs, + client, + filenames, + read_buffer + ); + + // Then the pending eviction pass is silently skipped: the stale file remains + fs.ClearSimulatedFailure(alpha_pending_dir); + REQUIRE(fs.IsFile(alpha_pending_prefix + std::to_string(old_ts))); + + // And no upload requests were made + REQUIRE(client.requests.empty()); + } + SECTION("M reject batch and continue processing W upload gets HTTP client error") { // Given a single registered feature with two eligible batches that can be read // and processed without error diff --git a/tests/mock/filesystem.cpp b/tests/mock/filesystem.cpp index a7736cf0..556d2fe0 100644 --- a/tests/mock/filesystem.cpp +++ b/tests/mock/filesystem.cpp @@ -485,6 +485,20 @@ FilesystemResult MockFilesystem::Close(PlatformFileHandle handle) { return FilesystemResult::OK; } +IFilesystem::GetFileSizeResult MockFilesystem::GetFileSize(const PlatformPath& path) { + const std::string normalized_path = NormalizePath(path); + std::lock_guard lock(_mutex); + + const auto found = _files.find(normalized_path); + if (found == _files.end()) { + return {FilesystemResult::DoesNotExist, 0}; + } + if (auto status = HasSimulatedFailure(found->second, FailureFlags::Stat)) { + return {*status, 0}; + } + return {FilesystemResult::OK, found->second.data.size()}; +} + FilesystemResult MockFilesystem::Delete(const PlatformPath& path) { const std::string normalized_path = NormalizePath(path); std::lock_guard lock(_mutex); @@ -931,7 +945,7 @@ bool MockFilesystem::IsFileLocked(std::string_view path) { if (file != _files.end()) { return file->second.advisory_lock_holder != INVALID_FILE_HANDLE; } - return ""; + return false; } std::string MockFilesystem::Cat(std::string_view path) { diff --git a/tests/mock/filesystem.hpp b/tests/mock/filesystem.hpp index 7795b39a..11183494 100644 --- a/tests/mock/filesystem.hpp +++ b/tests/mock/filesystem.hpp @@ -32,6 +32,7 @@ class MockFilesystem : public impl::IFilesystem { * at (or, in some cases, one level beneath) the given path. */ enum class FailureFlags : uint8_t { + Stat = (1 << 0), // GetFileSize will fail for the target file Mkdir = (1 << 1), // CreateDirectory will fail with target dir as parent Ls = (1 << 2), // ListFiles and ListSubdirectories will fail in target dir Open = (1 << 3), // OpenForRead and OpenForWrite will fail on file; in target dir @@ -131,6 +132,7 @@ class MockFilesystem : public impl::IFilesystem { ) override; ReadResult Read(impl::PlatformFileHandle handle, char* dst, size_t n) override; impl::FilesystemResult Close(impl::PlatformFileHandle handle) override; + GetFileSizeResult GetFileSize(const impl::PlatformPath& path) override; impl::FilesystemResult Delete(const impl::PlatformPath& path) override; impl::FilesystemResult DeleteDirectory(const impl::PlatformPath& path) override; impl::FilesystemResult Rename( From 242aa7e57658780b6197def1da71705c09a5683d Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 10:11:44 -0500 Subject: [PATCH 2/9] chore: Reformat --- src/datadog/impl/core/storage_write.cpp | 27 +++++++++++++------------ src/datadog/impl/core/upload_thread.cpp | 9 +++++---- tests/impl/core/storage_write_test.cpp | 4 +--- tests/impl/core/upload_thread_test.cpp | 4 +--- 4 files changed, 21 insertions(+), 23 deletions(-) diff --git a/src/datadog/impl/core/storage_write.cpp b/src/datadog/impl/core/storage_write.cpp index 5fb51f3d..da786973 100644 --- a/src/datadog/impl/core/storage_write.cpp +++ b/src/datadog/impl/core/storage_write.cpp @@ -434,8 +434,9 @@ bool BatchWriter::CacheKnownFilenames(const StoragePath& consent_dir_path) const } // Sort filenames for deterministic iteration in timestamp-name-order, then strip any - // non-numeric names (files not written by the SDK) so that both PurgeDirectoryIfNeeded - // and GetFilenameForNextWrite operate only on batch files we own. + // non-numeric names (files not written by the SDK) so that both + // PurgeDirectoryIfNeeded and GetFilenameForNextWrite operate only on batch files we + // own. std::sort(_last_known_filenames.begin(), _last_known_filenames.end()); _last_known_filenames.erase( std::remove_if( @@ -453,13 +454,13 @@ bool BatchWriter::CacheKnownFilenames(const StoragePath& consent_dir_path) const } void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) { - // _last_known_filenames is pre-populated, sorted, and already filtered to numeric-named - // (SDK-written) batch files by CacheKnownFilenames. + // _last_known_filenames is pre-populated, sorted, and already filtered to + // numeric-named (SDK-written) batch files by CacheKnownFilenames. - // Accumulate sizes for each file using a reusable path buffer. Files whose size cannot - // be determined are treated as zero: they don't push the running total toward the quota - // threshold, but remain eligible for deletion if the purge loop runs and reaches them - // (without contributing to total reduction when deleted). + // Accumulate sizes for each file using a reusable path buffer. Files whose size + // cannot be determined are treated as zero: they don't push the running total toward + // the quota threshold, but remain eligible for deletion if the purge loop runs and + // reaches them (without contributing to total reduction when deleted). FilesystemWrapper fsw(_fs); StoragePath path_buf; std::vector sizes; @@ -481,11 +482,11 @@ void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) { return; } - // Delete oldest files until we're within quota, erasing each from _last_known_filenames - // and the parallel sizes vector in lock-step so GetFilenameForNextWrite sees the - // post-purge state. Iterate by index rather than iterator so we can erase in-place: - // on a successful deletion we don't advance i (the next element slides into position); - // on failure we skip the file and increment. + // Delete oldest files until we're within quota, erasing each from + // _last_known_filenames and the parallel sizes vector in lock-step so + // GetFilenameForNextWrite sees the post-purge state. Iterate by index rather than + // iterator so we can erase in-place: on a successful deletion we don't advance i (the + // next element slides into position); on failure we skip the file and increment. size_t i = 0; while (i < _last_known_filenames.size() && total > _config.max_directory_size) { path_buf.MustSet(dir); diff --git a/src/datadog/impl/core/upload_thread.cpp b/src/datadog/impl/core/upload_thread.cpp index 889571b6..5b40507f 100644 --- a/src/datadog/impl/core/upload_thread.cpp +++ b/src/datadog/impl/core/upload_thread.cpp @@ -170,11 +170,12 @@ static _process_and_upload_batch_result _process_and_upload_batch( // relative to `now`. Returns nullopt for non-numeric filenames (non-SDK files that // should be left alone). Guards against underflow if the timestamp is in the future by // clamping the result to zero. -static std::optional batch_file_age(const std::string& filename, Timestamp now) { +static std::optional batch_file_age( + const std::string& filename, Timestamp now +) { uint64_t timestamp_ms{0}; - const auto parse_result = std::from_chars( - filename.data(), filename.data() + filename.size(), timestamp_ms - ); + const auto parse_result = + std::from_chars(filename.data(), filename.data() + filename.size(), timestamp_ms); if (parse_result.ec != std::errc{} || parse_result.ptr != filename.data() + filename.size()) { return std::nullopt; diff --git a/tests/impl/core/storage_write_test.cpp b/tests/impl/core/storage_write_test.cpp index 1d895b0d..e01e1128 100644 --- a/tests/impl/core/storage_write_test.cpp +++ b/tests/impl/core/storage_write_test.cpp @@ -941,9 +941,7 @@ TEST_CASE("BatchWriter", "[unit]") { REQUIRE(names[1] == "1700000000100"); } - SECTION( - "M treat GetFileSize failure as zero W file stat fails during quota check" - ) { + SECTION("M treat GetFileSize failure as zero W file stat fails during quota check") { // Given a batch writer with a 50-byte quota auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); config.max_directory_size = 50; diff --git a/tests/impl/core/upload_thread_test.cpp b/tests/impl/core/upload_thread_test.cpp index 8cff60ab..c5d42319 100644 --- a/tests/impl/core/upload_thread_test.cpp +++ b/tests/impl/core/upload_thread_test.cpp @@ -975,9 +975,7 @@ TEST_CASE("HandleUploadProc", "[unit]") { REQUIRE(client.requests.empty()); } - SECTION( - "M silently skip pending eviction W ListFiles fails on pending directory" - ) { + SECTION("M silently skip pending eviction W ListFiles fails on pending directory") { // Given an upload cycle with an 18-hour TTL clock.FreezeAtMilliseconds(1700000000000); MockHttpClient client; From 045b676d7215926211560fa7a650cfc2d1b8f16d Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 11:05:54 -0500 Subject: [PATCH 3/9] fix: Run upload-thread eviction check up-front This ensures that an early-out on the upload cycle due to failure to list files in the granted directory will not prevent deletion of outdated files in the pending directory. We still run the pending-directory eviction check once per upload cycle; it just runs at the start of the cycle (before upload attempts) rather than at the end (after upload attempts). --- src/datadog/impl/core/upload_thread.cpp | 88 +++++++++++++------------ 1 file changed, 46 insertions(+), 42 deletions(-) diff --git a/src/datadog/impl/core/upload_thread.cpp b/src/datadog/impl/core/upload_thread.cpp index 5b40507f..67222e8e 100644 --- a/src/datadog/impl/core/upload_thread.cpp +++ b/src/datadog/impl/core/upload_thread.cpp @@ -212,6 +212,52 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp // Prepare a FilesystemWrapper that will handle path encoding transparently FilesystemWrapper fsw(fs); + const Timestamp now = clock.Now(); + + // Evict stale files from the pending-consent directory. These accumulate when consent + // is set to Pending and never resolved. The upload thread never reads from this + // directory, so its only clean-up mechanism is this age-based eviction pass. + // + // This runs before the granted-directory listing so that a transient failure listing + // the granted directory does not prevent eviction of stale pending files. + const StoragePath& pending_dir_path = feature.storage->GetPendingPath(); + mut_filenames.clear(); + const auto pending_list_res = fsw.ListFiles(pending_dir_path.CStr(), mut_filenames); + if (pending_list_res == FilesystemResult::OK) { + std::sort(mut_filenames.begin(), mut_filenames.end()); + for (const std::string& pending_filename : mut_filenames) { + const auto file_age = batch_file_age(pending_filename, now); + if (!file_age) { + continue; + } + if (*file_age < config.max_file_age_for_read) { + // Files are sorted oldest-first; all subsequent files are also too young. + break; + } + + StoragePath pending_file_path; + pending_file_path.MustSet(pending_dir_path); + if (!pending_file_path.Append(pending_filename)) { + continue; + } + + const auto delete_res = fsw.Delete(pending_file_path.CStr()); + if (delete_res == FilesystemResult::OK) { + diagnostic_logger.Debug( + "Deleted outdated pending batch file", + {{"feature", feature.name}, {"filename", pending_filename}} + ); + } else { + diagnostic_logger.Warning( + "Failed to delete outdated pending batch file", + {{"feature", feature.name}, + {"filename", pending_filename}, + {"error", FilesystemResultStr(delete_res)}} + ); + } + } + } + // Retrieve a list of all filenames in the consent-granted event directory mut_filenames.clear(); const StoragePath& granted_dir_path = feature.storage->GetGrantedPath(); @@ -230,7 +276,6 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp _process_and_upload_batch_result last_batch_result{ _process_and_upload_batch_result::success }; - const Timestamp now = clock.Now(); for (const std::string& filename : mut_filenames) { const auto file_age = batch_file_age(filename, now); if (!file_age) { @@ -362,47 +407,6 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp } } - // Evict stale files from the pending-consent directory. These accumulate when consent - // is set to Pending and never resolved. The upload thread never reads from this - // directory, so its only clean-up mechanism is this age-based eviction pass. - const StoragePath& pending_dir_path = feature.storage->GetPendingPath(); - mut_filenames.clear(); - const auto pending_list_res = fsw.ListFiles(pending_dir_path.CStr(), mut_filenames); - if (pending_list_res == FilesystemResult::OK) { - std::sort(mut_filenames.begin(), mut_filenames.end()); - for (const std::string& pending_filename : mut_filenames) { - const auto file_age = batch_file_age(pending_filename, now); - if (!file_age) { - continue; - } - if (*file_age < config.max_file_age_for_read) { - // Files are sorted oldest-first; all subsequent files are also too young. - break; - } - - StoragePath pending_file_path; - pending_file_path.MustSet(pending_dir_path); - if (!pending_file_path.Append(pending_filename)) { - continue; - } - - const auto delete_res = fsw.Delete(pending_file_path.CStr()); - if (delete_res == FilesystemResult::OK) { - diagnostic_logger.Debug( - "Deleted outdated pending batch file", - {{"feature", feature.name}, {"filename", pending_filename}} - ); - } else { - diagnostic_logger.Warning( - "Failed to delete outdated pending batch file", - {{"feature", feature.name}, - {"filename", pending_filename}, - {"error", FilesystemResultStr(delete_res)}} - ); - } - } - } - if (num_uploads_attempted == 0) { diagnostic_logger.Debug( "Upload cycle finished with nothing to upload", {{"feature", feature.name}} From 917306cb0cc7c733314a52c8713b3e33501f662d Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 11:07:18 -0500 Subject: [PATCH 4/9] chore: Add note explaining that size limit is not inclusive of new file --- src/datadog/impl/core/storage_write.cpp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/datadog/impl/core/storage_write.cpp b/src/datadog/impl/core/storage_write.cpp index da786973..68df5cc5 100644 --- a/src/datadog/impl/core/storage_write.cpp +++ b/src/datadog/impl/core/storage_write.cpp @@ -478,6 +478,11 @@ void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) { total += file_size; } + // Note: this check covers only the files already on disk; it does not account for the + // new batch file that PrepareFileForNextWrite is about to create. The directory may + // therefore exceed max_directory_size by up to max_file_size after each rotation. + // This matches the iOS SDK's check-on-create design and the overshoot is bounded and + // transient, so it is accepted as a known limitation. if (total <= _config.max_directory_size) { return; } From 640075f578a1d67607892b310ef1f5466bc74f5b Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 11:37:06 -0500 Subject: [PATCH 5/9] fix: Document and handle race conditions on batch file eviction --- .../impl/core/storage/feature_event.cpp | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/src/datadog/impl/core/storage/feature_event.cpp b/src/datadog/impl/core/storage/feature_event.cpp index e5b19cb1..88802d54 100644 --- a/src/datadog/impl/core/storage/feature_event.cpp +++ b/src/datadog/impl/core/storage/feature_event.cpp @@ -170,6 +170,16 @@ bool FeatureEventStorage::MigratePendingBatchesToGranted() { // a file rename operation is atomic on both POSIX and Windows: the upload thread // will never see a "partially-moved" file. // + // Race with upload-thread age-based eviction: + // + // The upload thread runs an eviction pass over _pending_root (deleting files older + // than 18h) independently of this migration. If the upload thread deletes a file + // between our ListFiles call and our Rename call for that file, the Rename returns + // DoesNotExist. We treat that as "already gone; continue" rather than a fatal error: + // the file was expired anyway and would never have been uploaded. Aborting on + // DoesNotExist would strand all younger files that appear later in the sorted + // listing. + // // Additional considerations re: filename conflicts: // // - The files we're moving are named with timestamps indicating file creation time, @@ -261,6 +271,17 @@ bool FeatureEventStorage::MigratePendingBatchesToGranted() { continue; } + // If the rename failed because the source file no longer exists, the upload + // thread's age-based eviction has already removed it. The file was expired and + // would not have been uploaded; skip it and continue migrating the remaining files. + if (rename_res == FilesystemResult::DoesNotExist) { + _logger.Debug( + "Pending batch file already evicted before migration; skipping", + {{"path", src_file_path.Get()}} + ); + continue; + } + // If the move failed for any other reason, abort the migration operation _logger.Error( "Could not migrate batch file on consent change: rename failed", From 2f352646ebb8baab506788e42619fe3c915a62c1 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 11:57:44 -0500 Subject: [PATCH 6/9] fix: Check event size against lower of (max_event_size, max_file_size) --- src/datadog/impl/core/storage_write.cpp | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/datadog/impl/core/storage_write.cpp b/src/datadog/impl/core/storage_write.cpp index 68df5cc5..c79976a1 100644 --- a/src/datadog/impl/core/storage_write.cpp +++ b/src/datadog/impl/core/storage_write.cpp @@ -163,12 +163,17 @@ bool BatchWriter::FlushEvent( num_bytes += TLVBlockHeader::SIZE + event_metadata.size(); } - // If the encoded size of this single event (metadata + event, with headers) exceeds - // the per-event limit, reject it outright - if (num_bytes > _config.max_event_size) { + // Reject the event if it exceeds either the per-event cap or the batch file size + // limit. In the normal production config max_event_size < max_file_size, so the event + // cap is the binding constraint; taking the minimum preserves the file-size guard for + // configurations where max_file_size may be set smaller than max_event_size. + const size_t max_write_size = std::min(_config.max_event_size, _config.max_file_size); + if (num_bytes > max_write_size) { _diagnostic_logger.Error( - "Event dropped; size of single event exceeds max event size", - {{"event_size", num_bytes}, {"max_event_size", _config.max_event_size}} + "Event dropped; size of single event exceeds max write size", + {{"event_size", num_bytes}, + {"max_event_size", _config.max_event_size}, + {"max_file_size", _config.max_file_size}} ); return false; } From a7c1c2e8148ee840a2991dea90953d33a8c280e7 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 14:34:39 -0500 Subject: [PATCH 7/9] fix: In storage thread purge, handle already-deleted gracefully --- src/datadog/impl/core/storage_write.cpp | 16 +++++--- tests/impl/core/storage_write_test.cpp | 49 +++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/src/datadog/impl/core/storage_write.cpp b/src/datadog/impl/core/storage_write.cpp index c79976a1..03de174f 100644 --- a/src/datadog/impl/core/storage_write.cpp +++ b/src/datadog/impl/core/storage_write.cpp @@ -505,12 +505,18 @@ void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) { continue; } const auto delete_res = fsw.Delete(path_buf.CStr()); - if (delete_res == FilesystemResult::OK) { + // DoesNotExist means the upload thread already removed this file concurrently; + // treat it as a successful removal so total and _last_known_filenames stay + // consistent and we don't over-delete younger files to compensate. + if (delete_res == FilesystemResult::OK || + delete_res == FilesystemResult::DoesNotExist) { total -= sizes[i]; - _diagnostic_logger.Debug( - "Deleted batch file to enforce directory size quota", - {{"path", path_buf.Get()}} - ); + if (delete_res == FilesystemResult::OK) { + _diagnostic_logger.Debug( + "Deleted batch file to enforce directory size quota", + {{"path", path_buf.Get()}} + ); + } const auto idx = static_cast(i); _last_known_filenames.erase(_last_known_filenames.begin() + idx); sizes.erase(sizes.begin() + idx); diff --git a/tests/impl/core/storage_write_test.cpp b/tests/impl/core/storage_write_test.cpp index e01e1128..f25e38b6 100644 --- a/tests/impl/core/storage_write_test.cpp +++ b/tests/impl/core/storage_write_test.cpp @@ -972,6 +972,55 @@ TEST_CASE("BatchWriter", "[unit]") { REQUIRE(fs.IsFile(pending_prefix + "1700000000002")); } + SECTION( + "M not over-delete younger files W upload thread removes oldest file between " + "stat pass and delete pass" + ) { + // Given a batch writer with a directory quota of 50 bytes + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large); + config.max_directory_size = 50; + BatchWriter writer = init_writer(config); + + // And three existing batch files at t=0, t=1, t=2, each 40 bytes — total 120 bytes, + // well over the 50-byte quota + fs.Touch(pending_prefix + "1700000000000", std::string(40, 'x')); + fs.Touch(pending_prefix + "1700000000001", std::string(40, 'y')); + fs.Touch(pending_prefix + "1700000000002", std::string(40, 'z')); + + // The upload thread concurrently removes the oldest file between the stat pass and + // the deletion pass: simulate by making Delete return DoesNotExist for t=0 while + // leaving GetFileSize intact so the stat pass still sees its size (40 bytes). + // + // Note: the mock retains the file on disk even though Delete returns DoesNotExist; + // this mirrors the real scenario where the file is already gone externally but the + // mock cannot remove it via the simulated return-code path. + fs.SimulateFailure( + pending_prefix + "1700000000000", + FilesystemResult::DoesNotExist, + MockFilesystem::FailureFlags::Delete + ); + + // When a new write triggers the quota check + clock.TickMilliseconds(100); + REQUIRE(writer.HandleWrite("event-new", {}, false)); + + // Then the DoesNotExist result for t=0 is treated as a successful removal: its 40 + // bytes are subtracted, leaving 80 bytes > 50. The loop then removes t=1 (40 + // bytes), leaving 40 bytes ≤ 50 and stopping before t=2. Three files remain on + // disk: t=0 (physically present in mock despite DoesNotExist), t=2 (untouched), + // and the new batch file at t=100. + // + // Without the fix the DoesNotExist would be treated as a failure, keeping 120 bytes + // in total; the loop would then delete t=1 and t=2, leaving only t=0 and the new + // file — one more deletion than necessary. + std::vector names = fs.Ls(pending_dir_path); + std::sort(names.begin(), names.end()); + REQUIRE(names.size() == 3); + REQUIRE(names[0] == "1700000000000"); + REQUIRE(names[1] == "1700000000002"); + REQUIRE(names[2] == "1700000000100"); + } + SECTION("M write bypass event to granted dir W active file is warm in pending dir") { // Given a batch writer with consent Pending and a prior normal write that has // warmed up _active_file to a file in the pending directory From bbb7e733215fffb8e06489d0074fdd8c32226742 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 13 Jul 2026 22:40:23 -0500 Subject: [PATCH 8/9] fix: Tolerate DoesNotExist on upload-thread eviction race --- .../impl/core/storage/feature_event.cpp | 48 ++++++++++++------- tests/impl/core/storage_write_test.cpp | 38 +++++++++++++++ 2 files changed, 70 insertions(+), 16 deletions(-) diff --git a/src/datadog/impl/core/storage/feature_event.cpp b/src/datadog/impl/core/storage/feature_event.cpp index 88802d54..c08b2f3b 100644 --- a/src/datadog/impl/core/storage/feature_event.cpp +++ b/src/datadog/impl/core/storage/feature_event.cpp @@ -91,16 +91,21 @@ bool FeatureEventStorage::Initialize( bool FeatureEventStorage::DeletePendingBatches() { // Thread-safety/synchronization considerations: // - // 1. We only delete from _pending_root, which the upload thread never reads from, so - // we don't have to worry about contention with other threads - // 2. The storage thread performs deletion synchronously, i.e. there can be no file + // 1. The storage thread performs deletion synchronously, i.e. there can be no file // writes happening concurrently during deletion - // 3. When the storage thread _does_ perform writes, it does so atomically and closes + // 2. When the storage thread _does_ perform writes, it does so atomically and closes // the file, so there are no open handles to any of the files we're deleting - // 4. This function is called on the storage thread + // 3. This function is called on the storage thread // - // Therefore, we can safely assume that we have exclusive access to _pending_root - // during this function call. + // Race with upload-thread age-based eviction: + // + // The upload thread runs an eviction pass over _pending_root (deleting files older + // than 18h) independently of this deletion. If the upload thread deletes a file + // between our ListFiles call and our Delete call for that file, the Delete returns + // DoesNotExist. We treat that as "already gone; continue" rather than a fatal error: + // the file was expired anyway and would never have been uploaded. Aborting on + // DoesNotExist would strand all younger files that appear later in the sorted + // listing. // Use a FilesystemWrapper to handle path encoding transparently FilesystemWrapper fsw(_fs); @@ -137,20 +142,31 @@ bool FeatureEventStorage::DeletePendingBatches() { return false; } - // Attempt to delete the batch file, aborting on failure + // Attempt to delete the batch file const FilesystemResult delete_res = fsw.Delete(file_path.CStr()); - if (delete_res != FilesystemResult::OK) { - _logger.Error( - "Could not delete batch file on consent change", - {{"path", file_path.Get()}, {"error", FilesystemResultStr(delete_res)}} + if (delete_res == FilesystemResult::OK) { + _logger.Debug( + "Deleted batch file due to consent change", {{"path", file_path.Get()}} ); - return false; + continue; + } + + // If the file is already gone, the upload thread's eviction pass beat us to it. + // The file was expired and would not have been uploaded; skip it and continue. + if (delete_res == FilesystemResult::DoesNotExist) { + _logger.Debug( + "Pending batch file already evicted before consent-revocation cleanup; " + "skipping", + {{"path", file_path.Get()}} + ); + continue; } - // Batch file deleted successfully - _logger.Debug( - "Deleted batch file due to consent change", {{"path", file_path.Get()}} + _logger.Error( + "Could not delete batch file on consent change", + {{"path", file_path.Get()}, {"error", FilesystemResultStr(delete_res)}} ); + return false; } // Success: all batch files in pending directory deleted diff --git a/tests/impl/core/storage_write_test.cpp b/tests/impl/core/storage_write_test.cpp index f25e38b6..8c99b07d 100644 --- a/tests/impl/core/storage_write_test.cpp +++ b/tests/impl/core/storage_write_test.cpp @@ -541,6 +541,44 @@ TEST_CASE("BatchWriter", "[unit]") { REQUIRE(fs.Ls(granted_dir_path).size() == 0); } + SECTION( + "M delete remaining pending files W upload thread has already evicted one file " + "before consent-revocation cleanup runs" + ) { + // Given a BatchWriter that has written two batch files to the pending directory + auto config = BatchWriterConfig::FromBatchSize(BatchSize::Small); + BatchWriter writer = init_writer(config); + REQUIRE(writer.HandleWrite("event-0", {}, false)); + clock.Tick(std::chrono::seconds(60)); + REQUIRE(writer.HandleWrite("event-1", {}, false)); + + // Then we have two batch files in the pending directory + auto names = fs.Ls(pending_dir_path); + std::sort(names.begin(), names.end()); + REQUIRE(names.size() == 2); + + // When the upload thread concurrently evicts the oldest file between our ListFiles + // call and our Delete call: simulate this by making Delete return DoesNotExist for + // the oldest pending file + fs.SimulateFailure( + pending_prefix + names[0], + FilesystemResult::DoesNotExist, + MockFilesystem::FailureFlags::Delete + ); + + // And consent is revoked + const bool ok = writer.SetTrackingConsent(TrackingConsent::NotGranted); + + // Then the revocation succeeds despite the race: the DoesNotExist is treated as + // "already gone" and deletion continues to the younger file + REQUIRE(ok); + + // The younger file is deleted; the older file remains in the mock because + // SimulateFailure intercepted (but did not execute) its Delete call + names = fs.Ls(pending_dir_path); + REQUIRE(names.size() == 1); + } + SECTION( "M move all files from pending to granted W tracking consent changes to Granted" ) { From 1cd562f9e07c6b4f748f6a15453b846adfbf64ac Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Tue, 14 Jul 2026 08:07:20 -0500 Subject: [PATCH 9/9] fix: Continue upload cycle if batch deleted between list and open --- src/datadog/impl/core/upload_thread.cpp | 9 +++- tests/impl/core/upload_thread_test.cpp | 65 +++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/datadog/impl/core/upload_thread.cpp b/src/datadog/impl/core/upload_thread.cpp index 67222e8e..6ca33529 100644 --- a/src/datadog/impl/core/upload_thread.cpp +++ b/src/datadog/impl/core/upload_thread.cpp @@ -110,8 +110,13 @@ static _process_and_upload_batch_result _process_and_upload_batch( const bool hold_advisory_lock = false; auto open_result = fsw.OpenForRead(file_path.CStr(), hold_advisory_lock); if (open_result.value != FilesystemResult::OK) { - // If we failed to open the file for any reason, leave it in place and continue to - // the next file + if (open_result.value == FilesystemResult::DoesNotExist) { + // The file was removed by another agent (quota purge or age eviction) between + // the directory listing and this open. Treat it as a consumed batch so the + // upload cycle continues rather than backing off. + return _process_and_upload_batch_result::bad_batch; + } + // For any other open failure, leave the file in place and retry later return _process_and_upload_batch_result::retryable_failure; } diff --git a/tests/impl/core/upload_thread_test.cpp b/tests/impl/core/upload_thread_test.cpp index c5d42319..c03694f2 100644 --- a/tests/impl/core/upload_thread_test.cpp +++ b/tests/impl/core/upload_thread_test.cpp @@ -693,6 +693,71 @@ TEST_CASE("HandleUploadProc", "[unit]") { REQUIRE(delay_until_next_cycle == std::chrono::milliseconds(55000)); } + SECTION( + "M skip batch and continue processing W batch file is gone when opened" + " {race between quota purge and upload thread}" + ) { + // Given a single registered feature with three eligible batches, the second of + // which disappears between the directory listing and the open (simulating a + // concurrent quota purge) + clock.FreezeAtMilliseconds(1700000000000); + MockHttpClient client; + fs.Touch( + alpha_granted_prefix + "1699999955000", + MockTLVFile().AppendEvent("event-0").ToString() + ); + fs.Touch( + alpha_granted_prefix + "1699999956000", + MockTLVFile().AppendEvent("event-1").ToString() + ); + fs.Touch( + alpha_granted_prefix + "1699999957000", + MockTLVFile().AppendEvent("event-2").ToString() + ); + const auto [core_config, config] = init_config( + BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium + ); + auto features = register_feature(UploadFrequency::Average); + + // And a filesystem that will report DoesNotExist when the second file is opened + // (simulating the file being removed by the quota purge after listing) + fs.SimulateFailure( + alpha_granted_prefix + "1699999956000", + FilesystemResult::DoesNotExist, + MockFilesystem::FailureFlags::Open + ); + + // When we process uploads for that feature + std::vector filenames; + std::vector read_buffer; + Duration delay_until_next_cycle = Internal_HandleUploadProc( + logger, + config, + clock, + CreateFeatureId("ALFA"), + features, + fs, + client, + filenames, + read_buffer + ); + + // Then our first and third batches are uploaded (the missing second is skipped) + REQUIRE(alpha->reports.size() == 2); + REQUIRE(client.requests.size() == 2); + REQUIRE(client.requests[0].body == "event-0"); + REQUIRE(client.requests[1].body == "event-2"); + + // And the upload cycle did not abort early: all three files are gone from disk + // (the first and third were deleted after successful upload; the second was deleted + // by the attempted fsw.Delete after bad_batch, since only its Open was blocked) + REQUIRE(fs.Ls(alpha_granted_dir).empty()); + + // And this feature's upload delay is reduced because the cycle finished with a + // successful upload + REQUIRE(delay_until_next_cycle == std::chrono::seconds(10)); + } + SECTION("M reject batch and continue processing W unable to process batch") { // Given a single registered feature with three eligible batches, the second of // which does not contain valid TLV data