fix: stack-safe paging of all checkpoints and ADR 02 response shape

queryAllPages recursed through flatMap, which for strict effects (e.g.
Identity) runs the continuation right away, adding stack frames per page.
The continuation now hands the collected records over to a tail-recursive
loop when it runs synchronously, and recurses within the effect otherwise
(asynchronous or lazy effects), the hand-over being atomic.

ADR 02: a flow checkpoints page item is one checkpoint with its
measurements nested (processStartTime, measurements, partitioning), and
parent -> child edges cannot be reconstructed today (Bet 3).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: lsulak

adrs/02-consumption-pattern-for-audits.md

@@ -80,8 +80,8 @@ Three steps, all on **today's** v2 API:
    `PATCH .../ancestors` joins the new parent's flows, but its already existing descendants do not.)
 3. **Pull all measurements across the chain, newest first, paged.**
    `GET /api/v2/flows/{flowId}/checkpoints?limit=100&offset=0&include-properties=true`.
-   Each row = one measure of one checkpoint, **tagged with its partitioning**. That is the audit dataset:
-   measurements + their dataset relationships across the whole lineage chain.
+   Each item = one checkpoint with all its measurements nested, **tagged with its partitioning**. That is the audit
+   dataset: measurements + their dataset relationships across the whole lineage chain.
 
 ### Worked example (HTTP)
 
@@ -96,16 +96,16 @@ GET /api/v2/partitionings/2000000010448528/main-flow                        ->
 GET /api/v2/flows/3000000000123456/checkpoints?limit=100&offset=0&include-properties=true
 ```
 
-Each page item carries `checkpointStartTime`, `measureName`/`measurementValue`, and the `partitioning` it belongs
-to — so relationships come for free.
+Each page item is one checkpoint carrying its `processStartTime`, its `measurements` (each a `measure` - name and
+measured columns - with its `result`), and the `partitioning` it belongs to — so relationships come for free.
 
 ### "Last 2 months" today (no server change needed)
 
-Because results are latest-first, page step 3 and **stop at the first row older than the cutoff**:
+Because results are latest-first, page step 3 and **stop at the first checkpoint older than the cutoff**:
 
 ```
-keep the page while  checkpointStartTime >= now - 2 months
-stop as soon as a row is older (all later rows are older too)
+keep the page while  processStartTime >= now - 2 months
+stop as soon as a checkpoint is older (all later checkpoints are older too)
 ```
 
 This works on the current API. Its only cost is over-fetching whole pages and re-implementing the cutoff in every
@@ -318,7 +318,8 @@ are deferred.)*
   time). For late/backfilled reports the two can differ.
 - **Volume at scale.** A long-lived flow accumulates many checkpoints; the proposed index is what keeps a
   time-window audit query cheap.
-- **Explicit parent -> child edges.** Flow membership gives the downstream lineage *set*; if an audit needs the exact
-  edge list (who is parent of whom), that is reconstructed today via `get_partitioning_ancestors` per node. A
-  dedicated "flow graph" (nodes + edges) endpoint is a possible small future addition.
+- **Explicit parent -> child edges.** Flow membership gives the downstream lineage *set* only. If an audit needs the
+  exact edge list (who is parent of whom), it cannot be reconstructed today: `get_partitioning_ancestors` returns the
+  flow roots a partitioning belongs to, not its immediate parents (see §6). The edges have to be persisted first
+  (Bet 3); a dedicated "flow graph" (nodes + edges) endpoint can follow.
 - **Access control** for audit consumers (read-only scopes/authorization) is out of scope here.

reader/src/main/scala/za/co/absa/atum/reader/core/Reader.scala

@@ -23,10 +23,14 @@ import sttp.model.Uri
 import sttp.monad.MonadError
 import sttp.monad.syntax._
 import za.co.absa.atum.model.envelopes.SuccessResponse.PaginatedResponse
+import za.co.absa.atum.reader.core.Reader.{Detached, NextPage, PageHandOver, Pending}
 import za.co.absa.atum.reader.core.RequestResult._
 import za.co.absa.atum.reader.server.ServerConfig
 import za.co.absa.atum.reader.exceptions.RequestException.CirceError
 
+import java.util.concurrent.atomic.AtomicReference
+import scala.annotation.tailrec
+
 /**
  *  Reader is a base class for reading data from a remote server.
  *  @param serverConfig    - the configuration how to reach the Atum server
@@ -52,6 +56,9 @@ abstract class Reader[F[_]](implicit
   /**
    *  Queries the pages one after another, starting at offset 0, until the server reports there is no more data.
    *
+   *  The paging is stack-safe for any effect, including the strict ones (e.g. `Identity`), where `flatMap` runs its
+   *  continuation right away and a plain recursion would add stack frames with each page.
+   *
    *  @param pageSize  - the size of the page (record count) to query
    *  @param queryPage - function querying a page of the given size (limit) at the given offset
    *  @return          - the records of all the pages, in order, or the first error encountered
@@ -60,13 +67,33 @@ abstract class Reader[F[_]](implicit
     pageSize: Int,
     queryPage: (Int, Long) => F[RequestResult[PaginatedResponse[T]]]
   ): F[RequestResult[Seq[T]]] = {
+    // If the continuation of a page runs before `flatMap` returns (strict effect), it only hands the records collected
+    // so far over to the loop, which then queries the next page iteratively. Otherwise (asynchronous or lazy effect)
+    // the continuation queries the next page itself, the effect taking care of the stack. The hand-over is atomic, as
+    // the continuation of an asynchronous effect may run on another thread at the same time.
+    @tailrec
     def queryFrom(offset: Long, collected: Vector[T]): F[RequestResult[Seq[T]]] = {
-      queryPage(pageSize, offset).flatMap {
-        case Right(page) if page.pagination.hasMore => queryFrom(offset + pageSize, collected ++ page.data)
+      val handOver = new AtomicReference[PageHandOver[T]](Pending)
+      val result = queryPage(pageSize, offset).flatMap {
+        case Right(page) if page.pagination.hasMore =>
+          val nextCollected = collected ++ page.data
+          if (handOver.compareAndSet(Pending, new NextPage(nextCollected))) {
+            me.unit(RequestOK[Seq[T]](nextCollected)) // discarded, the loop continues with the next page
+          } else {
+            queryFromWithinEffect(offset + pageSize, nextCollected)
+          }
         case Right(page) => me.unit(RequestOK[Seq[T]](collected ++ page.data))
         case Left(error) => me.unit(RequestFail[Seq[T]](error))
       }
+      handOver.getAndSet(Detached) match {
+        case nextPage: NextPage[T] => queryFrom(offset + pageSize, nextPage.collected)
+        case _ => result
+      }
     }
+    // not a tail call, therefore kept apart for `queryFrom` to remain tail-recursive
+    def queryFromWithinEffect(offset: Long, collected: Vector[T]): F[RequestResult[Seq[T]]] =
+      queryFrom(offset, collected)
+
     queryFrom(0, Vector.empty)
   }
 
@@ -85,3 +112,10 @@ abstract class Reader[F[_]](implicit
     response.map(_.toRequestResult)
   }
 }
+
+object Reader {
+  private sealed class PageHandOver[+T]
+  private final class NextPage[+T](val collected: Vector[T]) extends PageHandOver[T]
+  private val Pending: PageHandOver[Nothing] = new PageHandOver
+  private val Detached: PageHandOver[Nothing] = new PageHandOver
+}

reader/src/test/scala/za/co/absa/atum/reader/core/ReaderUnitTests.scala

@@ -22,6 +22,8 @@ import sttp.client3.testing.SttpBackendStub
 import sttp.client3.{Identity, SttpBackend}
 import sttp.client3.monad.IdMonad
 import sttp.monad.MonadError
+import za.co.absa.atum.model.envelopes.Pagination
+import za.co.absa.atum.model.envelopes.SuccessResponse.PaginatedResponse
 import za.co.absa.atum.reader.core.RequestResult.{RequestFail, RequestOK, RequestResult}
 import za.co.absa.atum.reader.exceptions.RequestException.ParsingException
 import za.co.absa.atum.reader.server.ServerConfig
@@ -36,6 +38,13 @@ class ReaderUnitTests extends AnyFunSuiteLike {
     override def mapRequestResultF[I, O](requestResult: RequestResult[I], f: I => F[RequestResult[O]]): F[RequestResult[O]] = {
       super.mapRequestResultF(requestResult, f)
     }
+
+    override def queryAllPages[T](
+      pageSize: Int,
+      queryPage: (Int, Long) => F[RequestResult[PaginatedResponse[T]]]
+    ): F[RequestResult[Seq[T]]] = {
+      super.queryAllPages(pageSize, queryPage)
+    }
   }
 
   private implicit val serverConfig: ServerConfig = ServerConfig("http://localhost:8080")
@@ -57,4 +66,22 @@ class ReaderUnitTests extends AnyFunSuiteLike {
     val result = reader.mapRequestResultF(requestResult, fnc)
     assert(result == requestResult)
   }
+
+  test("All pages are queried, without exhausting the stack even when the effect is synchronous") {
+    val recordCount = 100000L
+    def queryPage(limit: Int, offset: Long): Identity[RequestResult[PaginatedResponse[Long]]] =
+      RequestOK(PaginatedResponse(Seq(offset), Pagination(limit, offset, hasMore = offset + limit < recordCount)))
+    val reader = new ReaderForTesting[Identity]
+    val result = reader.queryAllPages(1, queryPage)
+    assert(result == RequestOK(0L until recordCount))
+  }
+
+  test("Querying all pages stops at the first page failing") {
+    def queryPage(limit: Int, offset: Long): Identity[RequestResult[PaginatedResponse[Long]]] =
+      if (offset < 2) RequestOK(PaginatedResponse(Seq(offset), Pagination(limit, offset, hasMore = true)))
+      else RequestFail(ParsingException("Just a test", ""))
+    val reader = new ReaderForTesting[Identity]
+    val result = reader.queryAllPages(1, queryPage)
+    assert(result == RequestFail(ParsingException("Just a test", "")))
+  }
 }

reader/src/test/scala/za/co/absa/atum/reader/core/Reader_CatsIOUnitTests.scala

@@ -23,8 +23,10 @@ import sttp.client3.SttpBackend
 import sttp.client3.testing.SttpBackendStub
 import sttp.monad.{MonadAsyncError, MonadError}
 import za.co.absa.atum.model.dto.PartitionDTO
+import za.co.absa.atum.model.envelopes.Pagination
+import za.co.absa.atum.model.envelopes.SuccessResponse.PaginatedResponse
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions.JsonSerializationSyntax
-import za.co.absa.atum.reader.core.RequestResult.RequestResult
+import za.co.absa.atum.reader.core.RequestResult.{RequestOK, RequestResult}
 import za.co.absa.atum.reader.server.ServerConfig
 
 class Reader_CatsIOUnitTests extends AnyFunSuiteLike {
@@ -33,6 +35,10 @@ class Reader_CatsIOUnitTests extends AnyFunSuiteLike {
   private class ReaderForTest[F[_]](implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ev: MonadError[F])
     extends Reader {
     override def getQuery[R: Decoder](endpointUri: String, params: Map[String, String]): F[RequestResult[R]] = super.getQuery(endpointUri, params)
+    override def queryAllPages[T](
+      pageSize: Int,
+      queryPage: (Int, Long) => F[RequestResult[PaginatedResponse[T]]]
+    ): F[RequestResult[Seq[T]]] = super.queryAllPages(pageSize, queryPage)
   }
 
   test("Using Cats IO based backend") {
@@ -49,4 +55,19 @@ class Reader_CatsIOUnitTests extends AnyFunSuiteLike {
     assert(result == Right(partitionDTO))
   }
 
+  test("Querying all pages using Cats IO based backend, the query being repeatable") {
+    import cats.effect.IO
+    import za.co.absa.atum.reader.implicits.io.catsIOMonadError
+
+    implicit val server: SttpBackendStub[IO, Any] = SttpBackendStub[IO, Any](implicitly[MonadAsyncError[IO]])
+    val recordCount = 10000L
+    def queryPage(limit: Int, offset: Long): IO[RequestResult[PaginatedResponse[Long]]] =
+      IO(RequestOK(PaginatedResponse(Seq(offset), Pagination(limit, offset, hasMore = offset + limit < recordCount))))
+
+    val reader = new ReaderForTest
+    val query = reader.queryAllPages(1, queryPage)
+    assert(query.unsafeRunSync() == RequestOK(0L until recordCount))
+    assert(query.unsafeRunSync() == RequestOK(0L until recordCount))
+  }
+
 }

reader/src/test/scala/za/co/absa/atum/reader/core/Reader_FutureUnitTests.scala

@@ -22,10 +22,14 @@ import sttp.client3.SttpBackend
 import sttp.client3.testing.SttpBackendStub
 import sttp.monad.MonadError
 import za.co.absa.atum.model.dto.PartitionDTO
+import za.co.absa.atum.model.envelopes.Pagination
+import za.co.absa.atum.model.envelopes.SuccessResponse.PaginatedResponse
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions.JsonSerializationSyntax
-import za.co.absa.atum.reader.core.RequestResult.RequestResult
+import za.co.absa.atum.reader.core.RequestResult.{RequestFail, RequestOK, RequestResult}
+import za.co.absa.atum.reader.exceptions.RequestException.ParsingException
 import za.co.absa.atum.reader.server.ServerConfig
 
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.Duration
 import scala.concurrent.{Await, Future}
 
@@ -35,6 +39,10 @@ class Reader_FutureUnitTests extends AnyFunSuiteLike {
   private class ReaderForTest[F[_]](implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ev: MonadError[F])
     extends Reader {
     override def getQuery[R: Decoder](endpointUri: String, params: Map[String, String]): F[RequestResult[R]] = super.getQuery(endpointUri, params)
+    override def queryAllPages[T](
+      pageSize: Int,
+      queryPage: (Int, Long) => F[RequestResult[PaginatedResponse[T]]]
+    ): F[RequestResult[Seq[T]]] = super.queryAllPages(pageSize, queryPage)
   }
 
   test("Using Future based backend") {
@@ -50,4 +58,34 @@ class Reader_FutureUnitTests extends AnyFunSuiteLike {
     assert(result == Right(partitionDTO))
   }
 
+  test("Querying all pages using Future based backend") {
+    import za.co.absa.atum.reader.implicits.future.futureMonadError
+
+    implicit val server: SttpBackend[Future, Any] = SttpBackendStub.asynchronousFuture
+    val recordCount = 1000L
+    // alternating already completed and asynchronously completing pages
+    def queryPage(limit: Int, offset: Long): Future[RequestResult[PaginatedResponse[Long]]] = {
+      def page = RequestOK(PaginatedResponse(Seq(offset), Pagination(limit, offset, hasMore = offset + limit < recordCount)))
+      if (offset % 2 == 0) Future.successful(page) else Future(page)
+    }
+
+    val reader = new ReaderForTest
+    val result = Await.result(reader.queryAllPages(1, queryPage), Duration(10, "second"))
+    assert(result == RequestOK(0L until recordCount))
+  }
+
+  test("Querying all pages using Future based backend stops at the first page failing") {
+    import za.co.absa.atum.reader.implicits.future.futureMonadError
+
+    implicit val server: SttpBackend[Future, Any] = SttpBackendStub.asynchronousFuture
+    def queryPage(limit: Int, offset: Long): Future[RequestResult[PaginatedResponse[Long]]] = Future {
+      if (offset < 2) RequestOK(PaginatedResponse(Seq(offset), Pagination(limit, offset, hasMore = true)))
+      else RequestFail(ParsingException("Just a test", ""))
+    }
+
+    val reader = new ReaderForTest
+    val result = Await.result(reader.queryAllPages(1, queryPage), Duration(3, "second"))
+    assert(result == RequestFail(ParsingException("Just a test", "")))
+  }
+
 }