fix: Impose whole-directory size limit; apply 18h limit to pending dir

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
@@ -170,6 +186,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 +287,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",

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.
    *

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<size_t>(st.st_size)};
+  }
+
   FilesystemResult Delete(const PlatformPath& path) override {
     const int result = unlink(path.Get());
     if (result == 0) {

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<size_t>(
+        (static_cast<uint64_t>(data.nFileSizeHigh) << 32) | data.nFileSizeLow
+    );
+    return {FilesystemResult::OK, size};
+  }
+
   FilesystemResult Delete(const PlatformPath& path) override {
     const BOOL result = DeleteFileW(path.Get());
     if (result == 0) {

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;

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);

src/datadog/impl/core/storage_write.cpp

@@ -8,6 +8,7 @@
 
 #include <algorithm>
 #include <array>
+#include <cctype>
 #include <charconv>
 #include <chrono>
 #include <iostream>
@@ -162,12 +163,17 @@ 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) {
+  // 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 batch file 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;
   }
@@ -288,11 +294,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 +370,11 @@ bool BatchWriter::CanReuseFileForNextWrite(
 }
 
 std::optional<std::pair<uint64_t, std::string>> 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<char, MAX_UINT64_DECIMAL_DIGITS> buffer{0};
@@ -427,9 +438,96 @@ 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<size_t> 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;
+  }
+
+  // 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;
+  }
+
+  // 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());
+    // 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];
+      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<std::ptrdiff_t>(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