Add log cache log streaming support

[ZPascal]

Summary

Adds live log streaming via Log Cache — the Java equivalent of cf tail --follow.

Previously the only Log Cache read path was logsRecent, a single snapshot request. There was no way to continuously receive new log envelopes without polling manually. This change introduces a logsTail API across all three relevant modules.

---

Motivation

Cloud Foundry dropped the legacy Loggregator Doppler streaming endpoint (DopplerClient.stream()) in Loggregator ≥ 107.0 (CFD ≥ 24.3 / TAS ≥ 4.0). The CF CLI replaced it with a polling loop over the Log Cache /api/v1/read endpoint — that is what cf tail --follow does today. This PR brings the same capability to Java consumers.

---

Changes

Changes in detail

cloudfoundry-client — new TailLogsRequest and LogCacheClient.logsTail()

New _TailLogsRequest.java (Immutables @Value.Immutable):

Field	Type	Default	Description
sourceId	String	—	App / service GUID (required)
startTime	Long (nullable)	now − 5 s (ns)	Cursor start time in UNIX nanoseconds
envelopeTypes	List<EnvelopeType> (nullable)	all types	Envelope type filter
nameFilter	String (nullable)	none	Regex name filter (Log Cache ≥ 2.1.0)
pollInterval	Duration	250 ms	Back-off between polls when no new data

New method on LogCacheClient:

/**
 * Continuously polls Log Cache /api/v1/read and streams new Envelopes as they appear.
 * Equivalent to the Go logcache.Walk() API and `cf tail --follow`.
 * The Flux never completes on its own – cancel the subscription to stop streaming.
 */
Flux<Envelope> logsTail(TailLogsRequest request);

---

cloudfoundry-client-reactor — non-blocking polling implementation

ReactorLogCacheEndpoints.logsTail() mirrors the Go logcache.Walk() algorithm, fully non-blocking:

1. An AtomicLong cursor starts at startTime (or now − 5 s).
2. Flux.defer builds a fresh ReadRequest from the current cursor on every repetition and calls GET /api/v1/read/{sourceId}?start_time=cursor.
3. Envelopes are sorted ascending by timestamp; the cursor advances to lastTimestamp + 1; each envelope is emitted individually downstream.
4. When the batch is empty, repeatWhen inserts a Mono.delay(pollInterval) before the next poll. When envelopes arrive, the next poll starts immediately.
5. The Flux is infinite; the caller cancels the subscription to stop.

Transient network errors are logged and retried. HTTP errors (4xx/5xx) are propagated to the subscriber.

---

cloudfoundry-operations — Applications.logsTail()

DefaultApplications delegates to LogCacheClient:

@Override
public Flux<Envelope> logsTail(TailLogsRequest request) {
    return this.logCacheClient
            .flatMapMany(client -> client.logsTail(request))
            .transform(OperationsLogging.log("Tail Application Logs"))
            .checkpoint();
}

---

Tests

Test	What it verifies
logsTailLogCache	Happy path: a single LOG/OUT envelope is forwarded correctly
logsTailLogCacheMultipleEnvelopes	3 envelopes with types OUT → ERR → OUT are emitted in order
logsTailLogCacheError	A RuntimeException from the client propagates unchanged to the subscriber
logsTailLogCacheOutAndErrEnvelopes	Both stdout and stderr envelopes are forwarded without filtering

---

Usage

TailLogsRequest request = TailLogsRequest.builder()
        .sourceId(applicationGuid)
        .envelopeTypes(List.of(EnvelopeType.LOG))
        .pollInterval(Duration.ofMillis(250))
        .build();

logCacheClient.logsTail(request)
        .filter(e -> e.getLog() != null)
        .map(e -> e.getLog().getPayloadAsText())
        .subscribe(System.out::println);

Or via the Operations API:

cloudFoundryOperations.applications()
        .logsTail(TailLogsRequest.builder().sourceId(appGuid).build())
        .filter(e -> e.getLog() != null)
        .map(e -> e.getLog().getPayloadAsText())
        .subscribe(System.out::println);

---

Relation to existing API

Method	Transport	Completes?	Use case
DopplerClient.stream() ⚠️ deprecated	WebSocket / Doppler	Yes (server closes)	Legacy streaming (Loggregator < 107.0)
LogCacheClient.recentLogs()	HTTP GET (single)	Yes	Fetch last N log lines
LogCacheClient.logsTail() ✅ new	HTTP GET (polling loop)	Never	Live streaming (Loggregator ≥ 107.0)

---

Checklist

• _TailLogsRequest Immutables value object
• LogCacheClient.logsTail() interface method
• ReactorLogCacheEndpoints.logsTail() — non-blocking Reactor implementation
• _ReactorLogCacheClient.logsTail() — delegate override
• Applications.logsTail() — Operations API method
• DefaultApplications.logsTail() — implementation
• Unit tests in DefaultApplicationsTest (120 passing)
• Integration tests

[Kehrlann]

The API introduced in #1265 was specifically designed to transition seamlessly from Doppler to Logcache.

The idea was to:

1. Deprecate the old log(...) which used Doppler types and the Doppler API
2. Introduce a new API that is cache-system independent, put Doppler in the beginning but silently swap Logcache later on

We are fine doing this, because if users care about the underlying logger implementation, they should use DopplerClient or LogCacheClient directly ; "applications" is an implementation-agnostic wrapper.

#1348 was specifically built on top of this.
This PR should continue the work, and evolve logs to use LogCacheClient's new tailing API to stream logs.
We should not introduce new methods in Applications, nor expose LogCache specific APIs in Applications.

[Kehrlann]

This is a regression: this method's logs(... recent=true) worked for users that didn't have Doppler (the extreme majority), and now it doesn't work anymore.

[Kehrlann]

[LLM-assisted review]

Opus 5.5 flagged an issue here: LogCache is eventually consistent, meaning that logs may arrive in the cache out of order. So reading from T to now() (say T+5), and setting the next cursor to T+6 will miss any logs emitted between T and T+5 that arrive after T+5.

Golang has a trick, WalkWithDelay: it drops envelopes it considers "too new" (< 2s). It introduces some delay (2s by default), but mechanically reads those dropped envelopes in the next poll loop.

[Kehrlann]

I believe we're swallowing every every 4xx and 5xx through ErrorPayloadMappers#fallback and ending the stream.
Including rate limits (429) or indeed transient 503 as the logger warns below.

[Kehrlann]

Here we're piggybacking on fake empty responses to trigger a retry and introduce a delayed backoff. We should probably use Reactor's .retryWhen(), which gives us retry-spec, with fine grained controls over backoff.
The current interval is fine for backing off, but it lacks retry limit cap (maxCount in Go).

[Kehrlann]

I think this introduces a race condition should you have multiple subscribers. The problem is that the returned Flux has a closure over cursor.

var x = lcc.logsTail(request); // creates cursor1

x.doOnNext(...).subscribe(); // uses cursor1 to advance and mutates it
// some delays
x.doOnNext(...).subscribe(); // also uses cursor1 and competes with the previous sub

You probably want to rework your reactive pipeline to create the cursor on subscription.

Claude suggests something like:

Flux<Envelope> logsTail(TailLogsRequest request) {
    return Mono.fromSupplier(() -> initialCursor(request))   // evaluated per subscription
            .flatMap(cursor -> poll(request, cursor))
            .expand(page -> page.isEmpty()
                    ? Mono.delay(request.getPollInterval()).then(poll(request, page.cursor))
                    : poll(request, page.nextCursor()))
            .concatMapIterable(page -> page.envelopes);
}

private Mono<Page> poll(TailLogsRequest request, long cursor) {
    return read(toReadRequest(request, cursor))
            .map(response -> new Page(cursor, envelopes(response)));
}

private static final class Page {
    final long cursor;
    final List<Envelope> envelopes;

    Page(long cursor, List<Envelope> envelopes) { ... }

    boolean isEmpty() { return envelopes.isEmpty(); }

    long nextCursor() { return envelopes.get(envelopes.size() - 1).getTimestamp() + 1; }
}