fix(logs): address the #595 review — cancel the body, surface auth errors, poll for --since

packages/cli/src/cli/commands/project/logs.ts

@@ -1,3 +1,4 @@
+import { setTimeout as delay } from "node:timers/promises";
 import type { Logger } from "@base44-cli/logger";
 import type { Command } from "commander";
 import { Option } from "commander";
@@ -129,8 +130,6 @@ function writeFollowLine(entry: LogEntry, jsonMode: boolean): void {
   process.stdout.write(`${line}\n`);
 }
 
-const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
-
 function streamEventToLogEntry(event: StreamLogEvent): LogEntry {
   return {
     time: event.time,
@@ -254,6 +253,17 @@ async function followLogs(
     functions: parseFunctionNames(options.function),
     env: options.env,
   };
+  if (options.since) {
+    // A stream only carries what happens from now on, so a run that asked for
+    // the past polls from the start rather than opening one.
+    logger.warn(
+      "--since reads the past, so this run polls instead of streaming (lines may lag ~20-30s).",
+    );
+    return pollLogs(functionNames, options, availableFunctionNames, jsonMode, {
+      lastTime: "",
+      boundaryKeys: new Set(),
+    });
+  }
   const opened = await connectWhileTransientlyUnavailable(filters);
   if (opened.kind !== "stream") {
     logger.warn(
@@ -429,11 +439,6 @@ async function logsAction(
   }
 
   if (options.follow) {
-    if (options.since) {
-      throw new InvalidInputError(
-        "--since cannot be combined with --follow yet (the realtime stream starts from now).",
-      );
-    }
     if (options.until) {
       throw new InvalidInputError(
         "--until cannot be combined with --follow (a stream has no end).",

packages/cli/src/core/resources/function/stream-api.ts

@@ -97,7 +97,6 @@ const STREAM_SILENCE_TIMEOUT_MS = 60_000;
 interface StreamReader {
   read(): Promise<{ done: boolean; value?: Uint8Array }>;
   cancel(): Promise<unknown>;
-  releaseLock(): void;
 }
 
 async function readOrSilence(
@@ -123,18 +122,17 @@ async function* readLines(
   try {
     while (true) {
       const result = await readOrSilence(reader);
-      if (result === "silence") {
-        await reader.cancel();
-        return;
-      }
+      if (result === "silence") return;
       if (result.done || !result.value) return;
       buffered += decoder.decode(result.value, { stream: true });
       const lines = buffered.split("\n");
       buffered = lines.pop() ?? "";
       yield* lines;
     }
   } finally {
-    reader.releaseLock();
+    // Cancel, don't just unlock: an unconsumed body keeps its socket alive in
+    // the fetch pool, so a long tail's reconnects pile up connections.
+    await reader.cancel().catch(() => {});
   }
 }
 
@@ -173,20 +171,21 @@ export const isWorthReconnecting = (status: number) => status >= 500;
 export async function openLogStream(
   filters: LogStreamFilters,
 ): Promise<LogStreamAttempt> {
+  // Outside the try on purpose: a missing or unrefreshable token is a real
+  // error to surface, not a transient failure to retry for 15 seconds.
+  const url = buildStreamUrl(filters);
+  const headers = {
+    Accept: "text/event-stream",
+    ...(await buildStreamAuthHeaders()),
+  };
   const connectPhase = new AbortController();
   const connectTimer = setTimeout(
     () => connectPhase.abort(),
     STREAM_CONNECT_TIMEOUT_MS,
   );
   let response: Response;
   try {
-    response = await fetch(buildStreamUrl(filters), {
-      headers: {
-        Accept: "text/event-stream",
-        ...(await buildStreamAuthHeaders()),
-      },
-      signal: connectPhase.signal,
-    });
+    response = await fetch(url, { headers, signal: connectPhase.signal });
   } catch {
     return { kind: "transient" };
   } finally {

packages/cli/tests/cli/logs.spec.ts

@@ -14,6 +14,21 @@ import {
 } from "@/core/resources/function/index.js";
 import { fixture, setupCLITests } from "./testkit/index.js";
 
+async function waitForStderr(
+  handle: { readonly stderr: readonly string[] },
+  pattern: RegExp,
+  timeoutMs = 5000,
+) {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (pattern.test(handle.stderr.join(""))) return;
+    await new Promise((resolve) => setTimeout(resolve, 50));
+  }
+  throw new Error(
+    `Timed out waiting for stderr matching ${pattern}. stderr: ${handle.stderr.join("")}`,
+  );
+}
+
 function entry(time: string, message: string): LogEntry {
   return { time, level: "info", message, source: "fn" };
 }
@@ -424,20 +439,24 @@ describe("logs command", () => {
     t.expectResult(result).toContain("No production logs found");
   });
 
-  it("rejects --follow combined with --since", async () => {
+  it("polls without trying to stream when --follow is combined with --since", async () => {
     await t.givenLoggedInWithProject(fixture("basic"));
+    t.api.mockFunctionLogs("my-function", []);
 
-    const result = await t.run(
+    const handle = await t.runLive(
       "logs",
       "--function",
       "my-function",
       "--follow",
       "--since",
       "1h",
     );
+    await waitForStderr(handle, /polls instead of streaming/);
+    const result = await handle.stop();
 
-    t.expectResult(result).toFail();
-    t.expectResult(result).toContain("--since cannot be combined");
+    // The two stream-failure warnings would say "not available for this app" or
+    // "Could not reach" instead, so this line is proof the stream was skipped.
+    t.expectResult(result).toContain("--since reads the past");
   });
 
   it("rejects --follow combined with --order", async () => {