diff --git a/backend/cmd/ftester/main.go b/backend/cmd/ftester/main.go index 8d72197b..7830da32 100644 --- a/backend/cmd/ftester/main.go +++ b/backend/cmd/ftester/main.go @@ -63,21 +63,11 @@ func main() { if err != nil && !errors.Is(err, obs.ErrNotConfigured) { log.Fatalf("Unable to create langfuse client: %v\n", err) } - defer func() { - if lfclient != nil { - lfclient.ForceFlush(context.Background()) - } - }() otelclient, err := obs.NewTelemetryClient(ctx, cfg) if err != nil && !errors.Is(err, obs.ErrNotConfigured) { log.Fatalf("Unable to create telemetry client: %v\n", err) } - defer func() { - if otelclient != nil { - otelclient.ForceFlush(context.Background()) - } - }() obs.InitObserver(ctx, lfclient, otelclient, []logrus.Level{ logrus.DebugLevel, @@ -85,6 +75,13 @@ func main() { logrus.WarnLevel, logrus.ErrorLevel, }) + // Drain telemetry on exit, bounded — an unreachable collector must not hang + // the tester on the SDK's per-provider timeouts. + defer func() { + drainCtx, cancelDrain := context.WithTimeout(context.Background(), 5*time.Second) + defer cancelDrain() + _ = obs.Observer.Drain(drainCtx) + }() // Initialize database connection db, err := sql.Open("postgres", cfg.DatabaseURL) diff --git a/backend/cmd/pentagi/main.go b/backend/cmd/pentagi/main.go index 435a78ac..3ba824ab 100644 --- a/backend/cmd/pentagi/main.go +++ b/backend/cmd/pentagi/main.go @@ -57,8 +57,8 @@ func main() { logrus.SetLevel(logrus.InfoLevel) } - // Telemetry is optional: degrade to a no-op observer on init failure instead - // of killing the process, so an unreachable collector can't take the app down. + // Telemetry is optional — degrade to a no-op observer on init failure so an + // unreachable collector can't take the app down. lfclient, err := obs.NewLangfuseClient(ctx, cfg) if err != nil && !errors.Is(err, obs.ErrNotConfigured) { logrus.WithError(err).Warn("langfuse telemetry disabled: client init failed") diff --git a/backend/pkg/docker/client.go b/backend/pkg/docker/client.go index 067d2caa..62d01c44 100644 --- a/backend/pkg/docker/client.go +++ b/backend/pkg/docker/client.go @@ -705,24 +705,34 @@ func demuxExecStdout(r io.Reader, maxStdout int) ([]byte, error) { header := make([]byte, 8) for { if _, err := io.ReadFull(r, header); err != nil { - if err == io.EOF || err == io.ErrUnexpectedEOF { - break + if err == io.EOF { + break // clean end at a frame boundary } - return nil, err + // A header cut short (ErrUnexpectedEOF) means the stream was truncated + // mid-frame — the listing is incomplete, so fail rather than silently + // dropping the tail. + return nil, fmt.Errorf("truncated exec stream: %w", err) } size := int64(binary.BigEndian.Uint32(header[4:8])) if size == 0 { continue } - sink := io.Writer(io.Discard) - if header[0] == 1 { // stdout - sink = &stdout - } - if _, err := io.CopyN(sink, r, size); err != nil { - return nil, err - } - if stdout.Len() > maxStdout { - return nil, fmt.Errorf("listing output exceeded %d bytes", maxStdout) + switch header[0] { + case 1: // stdout + if _, err := io.CopyN(&stdout, r, size); err != nil { + return nil, err + } + if stdout.Len() > maxStdout { + return nil, fmt.Errorf("listing output exceeded %d bytes", maxStdout) + } + case 3: // systemerr — a daemon-level error injected mid-stream; surface it + var msg bytes.Buffer + _, _ = io.CopyN(&msg, r, size) + return nil, fmt.Errorf("docker exec systemerr: %s", strings.TrimSpace(msg.String())) + default: // stderr and anything else — discard + if _, err := io.CopyN(io.Discard, r, size); err != nil { + return nil, err + } } } return stdout.Bytes(), nil diff --git a/backend/pkg/docker/client_listing_test.go b/backend/pkg/docker/client_listing_test.go index b3b57f69..258edcf1 100644 --- a/backend/pkg/docker/client_listing_test.go +++ b/backend/pkg/docker/client_listing_test.go @@ -53,3 +53,17 @@ func TestDemuxExecStdout_StdoutOnly_AndByteCap(t *testing.T) { t.Fatalf("want cap error, got %v", err) } } + +func TestDemuxExecStdout_TruncatedAndSystemerr(t *testing.T) { + // a header cut short mid-frame must error, not silently drop the tail + torn := append(listingFrame(1, "a.txt\x00"), 0x01, 0x00, 0x00) // 3 stray header bytes + if _, err := demuxExecStdout(bytes.NewReader(torn), 1<<20); err == nil || !strings.Contains(err.Error(), "truncated") { + t.Fatalf("want truncated-stream error, got %v", err) + } + + // a systemerr (stream id 3) daemon error must surface, not be discarded + sys := append(listingFrame(1, "ok"), listingFrame(3, "daemon connection reset")...) + if _, err := demuxExecStdout(bytes.NewReader(sys), 1<<20); err == nil || !strings.Contains(err.Error(), "systemerr") { + t.Fatalf("want systemerr, got %v", err) + } +} diff --git a/backend/pkg/server/services/flow_files.go b/backend/pkg/server/services/flow_files.go index e5b8ef10..59baaa59 100644 --- a/backend/pkg/server/services/flow_files.go +++ b/backend/pkg/server/services/flow_files.go @@ -1108,6 +1108,15 @@ func (s *FlowFileService) GetFlowContainerFiles(c *gin.Context) { return allFiles[i].ModifiedAt.Before(allFiles[j].ModifiedAt) }) + // Failures are sorted too so the dialog's first-N skipped-entries preview is + // deterministic rather than in raw find/readdir order. + sort.Slice(allFailures, func(i, j int) bool { + if allFailures[i].Path != allFailures[j].Path { + return allFailures[i].Path < allFailures[j].Path + } + return allFailures[i].Name < allFailures[j].Name + }) + // Backward-compat: when exactly one container path was queried, echo it back // as Path so existing callers receive the same field value as before. responsePath := "" @@ -1115,27 +1124,22 @@ func (s *FlowFileService) GetFlowContainerFiles(c *gin.Context) { responsePath = containerPaths[0] } - // A per-entry stat failure (dangling symlink, a file removed mid-listing, - // transient /proc entries) degrades the listing but must not blank it: the - // readable files are returned and each skipped entry is logged and carried - // back in Failures so an operator can tell what and how much was skipped. + // A per-entry stat failure (a file removed mid-listing, a transient /proc pid, + // a symlink loop) degrades the listing but must not blank it: the readable + // files are returned and each skipped entry is carried back in Failures so the + // caller can tell what and how much was skipped. if len(allFailures) > 0 { log := logger.FromContext(c) - const maxLoggedFailures = 20 - for i, fe := range allFailures { - if i >= maxLoggedFailures { - log.WithFields(map[string]any{ - "flow_id": flowID, - "suppressed": len(allFailures) - maxLoggedFailures, - }).Warn("additional container entries skipped (log capped)") - break - } + for _, fe := range allFailures { + // Detail is already in the Failures response, so log per-entry at Debug, + // not Warn — this endpoint is hit on every navigation. Quote the names so + // control bytes in a hostile filename can't inject into the log line. log.WithFields(map[string]any{ "flow_id": flowID, - "entry": fe.Name, - "entry_path": fe.Path, + "entry": strconv.Quote(fe.Name), + "entry_path": strconv.Quote(fe.Path), "error": rawFailureMessages[fe.Path], - }).Warn("container entry skipped: could not stat") + }).Debug("container entry skipped: could not stat") } log.WithFields(map[string]any{ "flow_id": flowID,