ktf · Nov 5, 2026 · Nov 5, 2026
@@ -13,6 +13,7 @@
 # ROOT, at build or run time: see the check at the end of this file.
 o2_add_library(AODNative
                SOURCES src/ByteSource.cxx
+                       src/FailoverByteSource.cxx
                        src/LocalByteSource.cxx
                        src/Container.cxx
                        src/Objects.cxx
@@ -54,7 +55,8 @@ add_executable(o2-test-framework-aod-native
                test/test_Codecs.cxx
                test/test_Objects.cxx
                test/test_Decoder.cxx
-               test/test_ReadPlan.cxx)
+               test/test_ReadPlan.cxx
+               test/test_Failover.cxx)
 target_include_directories(o2-test-framework-aod-native PRIVATE ${CMAKE_CURRENT_LIST_DIR}/src)
 target_link_libraries(o2-test-framework-aod-native PRIVATE O2::AODNative O2::Catch2 PkgConfig::zstd PkgConfig::lz4 PkgConfig::lzma ZLIB::ZLIB)
 

@@ -229,10 +229,20 @@ ROOT's 50 MB TTreeCache. With the macOS allocator cache on, max RSS is higher
 
 `XRootDByteSource` (root://, roots://, xroot://, xroots://), built when O2 finds
 XRootD (optional in O2); otherwise those URLs get an IoError, so `auto` falls
-back. Links only libXrdCl. `readvInto` sends a whole read plan at once: ranges
-up to 2097136 bytes go in vectored reads of at most 1024 chunks, larger ones
-in plain reads of at most 64 MiB, all asynchronous, then waits for all of them;
-a short or failed reply is an IoError. `readCalls()` counts requests sent.
+back. Links only libXrdCl. After opening it asks the data server for its
+kXR_readv limits (`readv_ior_max`, `readv_iov_max`), as TNetXNGFile does,
+keeping XrdCl's defaults (2097136 bytes, 1024 chunks) when it does not answer.
+`readvInto` sends ranges up to the chunk limit in vectored reads and larger ones
+in plain reads of at most 8 MiB, asynchronously, in rounds of at most
+`Options::remoteInFlightBytes` (32 MiB), waiting for each round; a short or
+failed reply is an IoError. `readCalls()` counts requests sent.
+
+Until 2026-11-05 a whole plan went out at once, with plain reads of up to
+64 MiB. On the Grid, at ORNL reading Poznan EOS (slow transatlantic link), the
+native reader then failed 2-6x more often than ROOT, mostly with "Operation
+expired": requests sharing the link took longer than XRD_REQUESTTIMEOUT (150 s
+in the ALICE environment). Smaller requests and the in-flight limit keep each
+request well within it.
 
 Tested against a local server (Homebrew xrootd 6.2.0, `all.export ~/src/data
 r/o`): byte-identical with both readers over root:// on hy_3993289/0065,
@@ -319,9 +329,16 @@ in DPL (analysis-histograms, LHC24ar) 380 vs 319, with identical results.
 sends `access read <lfn>` through `gGrid`, as `TJAlienFile::Open` does, and
 turns each replica into `root://<se>//<pfn>?&authz=<envelope>`. Files inside
 an archive (`#member`, on the LFN or the replica) are refused, so `auto` falls
-back to ROOT. `openNativeAOD()` tries the replicas in the order AliEn proposes,
-moving on after an IoError only (a format problem is the same on every
-replica); DPL's reader and the comparison tool both open files through it.
+back to ROOT. `openNativeAOD()` reads the replicas, in the order AliEn
+proposes, through `ByteSource::openReplicas()` (`FailoverByteSource`): the
+first replica which opens is read; a read which fails with an IoError, at any
+point of the file, is retried on that replica, reopened, up to
+`Options::readRetries` (2) times per replica with a doubling delay from
+`retryDelay` (2 s), and then continues on the next replica of the same size.
+Only IoErrors are retried: a format problem is the same on every replica. Each
+retry and change of replica is logged. ROOT only moves to another replica when
+opening fails, so this makes the native reader more robust than ROOT on flaky
+storage. DPL's reader and the comparison tool both open files through it.
 The file keeps its LFN as name: what DataInputDirector compares, what goes
 with derived output, and what monitoring reports, with `se=` and
 `open_time=` as for TJAlienFile. `XRootDByteSource` uses the full URL only to

@@ -14,9 +14,11 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <span>
 #include <string>
+#include <vector>
 
 namespace o2::aod::native
 {
@@ -46,6 +48,23 @@ struct Options {
   size_t streamWindowBytes = 2 * 1024 * 1024;
   HttpVersion httpVersion = HttpVersion::Auto;
   int httpConnections = 4;
+  /// For a source opened with ByteSource::openReplicas: how many times, at
+  /// most, failed reads are retried on a replica (reopened) during the life of
+  /// the file before moving on to the next replica, and the seconds waited
+  /// before the first retry (doubled for each further one).
+  int readRetries = 2;
+  double retryDelay = 2.0;
+  /// Most bytes of one read plan in flight at once on a remote file, so that
+  /// each request finishes within the server timeout on a slow link (XRootD
+  /// expires requests after XRD_REQUESTTIMEOUT, 150 s in the ALICE setup).
+  size_t remoteInFlightBytes = 32 * 1024 * 1024;
+};
+
+/// One copy of a file, for ByteSource::openReplicas: where it is, and a label
+/// (e.g. the storage element) to report which one is being read.
+struct Replica {
+  std::string url;
+  std::string label;
 };
 
 struct ReadRequest {
@@ -77,7 +96,15 @@ class ByteSource
   virtual uint64_t bytesRead() const = 0;
   virtual uint64_t readCalls() const = 0;
   virtual std::string url() const = 0;
+  /// The label of the replica being read, for a source opened with openReplicas().
+  virtual std::string replicaLabel() const { return {}; }
   static std::unique_ptr<ByteSource> open(std::string url, const Options&);
+  /// Opens the first replica which opens, and on a read error retries it
+  /// (Options::readRetries times, reopened) and then moves on to the next
+  /// replica of the same size. @a log, if given, is told about every retry
+  /// and every change of replica. Throws IoError when no replica is left.
+  static std::unique_ptr<ByteSource> openReplicas(std::vector<Replica> replicas, const Options&,
+                                                  std::function<void(std::string const&)> log = {});
 };
 
 } // namespace o2::aod::native

@@ -11,6 +11,7 @@
 
 #include "AODNative/ByteSource.h"
 #include "AODNative/Errors.h"
+#include "FailoverByteSource.h"
 #include "LocalByteSource.h"
 #ifdef AODNATIVE_WITH_XROOTD
 #include "XRootDByteSource.h"
@@ -59,4 +60,10 @@ std::unique_ptr<ByteSource> ByteSource::open(std::string url, const Options& opt
   return std::make_unique<LocalByteSource>(std::move(url), options);
 }
 
+std::unique_ptr<ByteSource> ByteSource::openReplicas(std::vector<Replica> replicas, const Options& options,
+                                                     std::function<void(std::string const&)> log)
+{
+  return std::make_unique<FailoverByteSource>(std::move(replicas), options, &ByteSource::open, std::move(log));
+}
+
 } // namespace o2::aod::native
@@ -0,0 +1,151 @@
+// Copyright 2019-2026 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#include "FailoverByteSource.h"
+#include "AODNative/Errors.h"
+#include <chrono>
+#include <cmath>
+#include <thread>
+
+namespace o2::aod::native
+{
+
+namespace
+{
+std::string labelOf(Replica const& r)
+{
+  return r.label.empty() ? r.url.substr(0, r.url.find('?')) : r.label;
+}
+} // namespace
+
+FailoverByteSource::FailoverByteSource(std::vector<Replica> replicas, Options const& options, Opener opener, Log log)
+  : mReplicas(std::move(replicas)), mOptions(options), mOpener(std::move(opener)), mLog(std::move(log))
+{
+  if (mReplicas.empty()) {
+    throw IoError("(no replica)", "no replica to open");
+  }
+  for (mIndex = 0; mIndex < mReplicas.size(); ++mIndex) {
+    try {
+      mSource = openReplica(mIndex);
+      mSize = mSource->size();
+      mRemote = mSource->isRemote();
+      return;
+    } catch (IoError const& e) {
+      report("Could not open replica on " + labelOf(mReplicas[mIndex]) + ": " + e.what());
+      mErrors += " " + labelOf(mReplicas[mIndex]) + ": " + e.what() + ";";
+    }
+  }
+  throw IoError(labelOf(mReplicas.front()), "no replica could be opened:" + mErrors);
+}
+
+std::shared_ptr<ByteSource> FailoverByteSource::openReplica(size_t index)
+{
+  std::shared_ptr<ByteSource> source = mOpener(mReplicas[index].url, mOptions);
+  if (mSource && source->size() != mSize) {
+    throw IoError(source->url(), "size " + std::to_string(source->size()) + " differs from the " + std::to_string(mSize) + " bytes of the replica read so far");
+  }
+  return source;
+}
+
+std::shared_ptr<ByteSource> FailoverByteSource::current() const
+{
+  std::lock_guard lock{mMutex};
+  return mSource;
+}
+
+std::string FailoverByteSource::replicaLabel() const
+{
+  std::lock_guard lock{mMutex};
+  return mReplicas[mIndex].label;
+}
+
+uint64_t FailoverByteSource::bytesRead() const
+{
+  return mRetiredBytes.load(std::memory_order_relaxed) + current()->bytesRead();
+}
+
+uint64_t FailoverByteSource::readCalls() const
+{
+  return mRetiredCalls.load(std::memory_order_relaxed) + current()->readCalls();
+}
+
+void FailoverByteSource::report(std::string const& message) const
+{
+  if (mLog) {
+    mLog(message);
+  }
+}
+
+void FailoverByteSource::readInto(uint64_t off, size_t len, void* dst)
+{
+  withFailover([&](ByteSource& s) { s.readInto(off, len, dst); });
+}
+
+void FailoverByteSource::readvInto(std::span<ReadRequest> requests)
+{
+  withFailover([&](ByteSource& s) { s.readvInto(requests); });
+}
+
+void FailoverByteSource::withFailover(std::function<void(ByteSource&)> const& read)
+{
+  while (true) {
+    std::shared_ptr<ByteSource> source;
+    uint64_t generation;
+    {
+      std::lock_guard lock{mMutex};
+      source = mSource;
+      generation = mGeneration;
+    }
+    try {
+      read(*source);
+      return;
+    } catch (IoError const& e) {
+      std::lock_guard lock{mMutex};
+      if (generation != mGeneration) {
+        continue; // another read already replaced the replica: try the new one
+      }
+      mRetiredBytes.fetch_add(source->bytesRead(), std::memory_order_relaxed);
+      mRetiredCalls.fetch_add(source->readCalls(), std::memory_order_relaxed);
+      std::string error = e.what();
+      // Reopen the same replica, after a growing delay, or move on to the next
+      // one, until a replica opens. Other readers wait on mMutex meanwhile.
+      while (true) {
+        if (mAttempts < mOptions.readRetries) {
+          double delay = mOptions.retryDelay * std::pow(2., mAttempts);
+          ++mAttempts;
+          report("Read error on " + labelOf(mReplicas[mIndex]) + ", retry " + std::to_string(mAttempts) + " of " +
+                 std::to_string(mOptions.readRetries) + " in " + std::to_string((int)std::lround(delay)) + " s: " + error);
+          std::this_thread::sleep_for(std::chrono::duration<double>(delay));
+        } else {
+          mErrors += " " + labelOf(mReplicas[mIndex]) + ": " + error + ";";
+          if (mIndex + 1 >= mReplicas.size()) {
+            throw IoError(labelOf(mReplicas.front()), "no replica left to read from:" + mErrors);
+          }
+          ++mIndex;
+          mAttempts = 0;
+          report("Moving to replica on " + labelOf(mReplicas[mIndex]) + " after: " + error);
+        }
+        try {
+          mSource = openReplica(mIndex);
+          ++mGeneration;
+          break;
+        } catch (IoError const& reopen) {
+          error = reopen.what();
+          if (mAttempts == 0) {
+            mAttempts = mOptions.readRetries; // a replica that does not open is not retried
+          }
+        }
+      }
+    }
+  }
+}
+
+} // namespace o2::aod::native
@@ -0,0 +1,78 @@
+// Copyright 2019-2026 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#ifndef O2_FRAMEWORK_AODNATIVE_FAILOVERBYTESOURCE_H_
+#define O2_FRAMEWORK_AODNATIVE_FAILOVERBYTESOURCE_H_
+
+#include "AODNative/ByteSource.h"
+#include <atomic>
+#include <functional>
+#include <mutex>
+
+namespace o2::aod::native
+{
+
+/// The replicas of one file, read as one ByteSource. A read which fails with
+/// an IoError is retried on the same replica, reopened, Options::readRetries
+/// times with a growing delay, and then on the next replica of the same size.
+/// Opening goes through the replicas in order too. Reads already in flight on
+/// a replica that is being replaced finish there, or fail and are retried.
+class FailoverByteSource : public ByteSource
+{
+ public:
+  using Opener = std::function<std::unique_ptr<ByteSource>(std::string const& url, Options const&)>;
+  using Log = std::function<void(std::string const&)>;
+
+  /// Throws IoError when no replica opens.
+  FailoverByteSource(std::vector<Replica> replicas, Options const& options, Opener opener, Log log);
+
+  uint64_t size() const override { return mSize; }
+  void readInto(uint64_t off, size_t len, void* dst) override;
+  void readvInto(std::span<ReadRequest> requests) override;
+  std::span<const uint8_t> tryMap(uint64_t off, size_t len) override { return current()->tryMap(off, len); }
+  bool canMap() const override { return current()->canMap(); }
+  bool isRemote() const override { return mRemote; }
+  void willNeed(uint64_t off, size_t len) override { current()->willNeed(off, len); }
+  void dontNeed(uint64_t off, size_t len) override { current()->dontNeed(off, len); }
+  uint64_t bytesRead() const override;
+  uint64_t readCalls() const override;
+  std::string url() const override { return current()->url(); }
+  std::string replicaLabel() const override;
+
+ private:
+  std::shared_ptr<ByteSource> current() const;
+  /// Runs @a read on the current replica, retrying and failing over on IoError.
+  void withFailover(std::function<void(ByteSource&)> const& read);
+  /// Opens replica @a index; throws IoError, also when its size differs.
+  std::shared_ptr<ByteSource> openReplica(size_t index);
+  void report(std::string const& message) const;
+
+  std::vector<Replica> mReplicas;
+  Options mOptions;
+  Opener mOpener;
+  Log mLog;
+
+  mutable std::mutex mMutex;
+  std::shared_ptr<ByteSource> mSource; // guarded by mMutex
+  size_t mIndex = 0;                   // guarded by mMutex
+  uint64_t mGeneration = 0;            // guarded by mMutex, bumped at every reopen
+  int mAttempts = 0;                   // retries done on the current replica, guarded by mMutex
+  std::string mErrors;                 // what went wrong on the replicas left behind
+
+  uint64_t mSize = 0;
+  bool mRemote = false;
+  std::atomic<uint64_t> mRetiredBytes = 0;
+  std::atomic<uint64_t> mRetiredCalls = 0;
+};
+
+} // namespace o2::aod::native
+
+#endif // O2_FRAMEWORK_AODNATIVE_FAILOVERBYTESOURCE_H_