Repository navigation
fix(persis): release provider sessions owned by foreach body steps - #3364
RyoKusnadi wants to merge 2 commits into
Conversation
Run cleanup collected provider sessions only from the run status nodes, so a session opened by a foreach body step was never released. Read the sessions from the foreach item records as well. Signed-off-by: Ryo Kusnadi <ryokusnadi@gmail.com>
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the
 |
| Layer / File(s) | Summary |
|---|---|
Discover foreach body sessions internal/persis/dagrun_repository.go |
withForeachBodyNodes scans matching item-status files and adds nodes for body steps with agent sessions. It skips directory-walk, file-read, and JSON-decoding errors. |
Include sessions in cleanup internal/persis/dagrun_repository.go, internal/persis/dagrun_repository_test.go |
Root and child run resource collection use the expanded node set. A test verifies that removal enqueues the foreach body session’s resource details. |
Priority:
Estimated code review effort: 2 (Simple) | ~10 minutes
Severity of issue fixed: Medium
Possibly related PRs
- dagucloud/dagu#3360: That PR stores
AgentSessionin foreach body-step status records; this change reads those records during cleanup.
Merge Risk: 
9d988
-
The change adds foreach body-step sessions to the existing run-removal cleanup, matching how top-level step sessions are handled. No merge-blocking risk remains.
Pre-merge checks: 
-
Check name Status Explanation Resolution Docstring Coverage Docstring coverage is 50.00% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Documentation-only edits and deletions within surv… Write docstrings for the functions missing them to satisfy the coverage threshold. -
Check name Status Explanation Title check The title clearly and concisely describes the main change: releasing provider sessions owned by foreach body steps. Description check The description explains the issue, implementation, linked issue, added test, and validation results. It does not use every template heading or include the checklist, but it provides the critical info… Linked Issues check Issue #3362requires cleanup to collect provider sessions owned by foreach body steps.agentSessionResourcesnow augments root and child run nodes with sessions read from foreach itemstatus.json…Out of Scope Changes check The changes are limited to foreach body session collection in internal/persis/dagrun_repository.goand a focused repository test. The implementation supports the linked cleanup behavior and does not…Full details: Description check
-
Explanation
The description explains the issue, implementation, linked issue, added test, and validation results. It does not use every template heading or include the checklist, but it provides the critical information needed to review the change.
Full details: Linked Issues check
-
Explanation
Issue
#3362requires cleanup to collect provider sessions owned by foreach body steps.agentSessionResourcesnow augments root and child run nodes with sessions read from foreach itemstatus.jsonrecords.filepath.WalkDiralso reaches nested foreach item records. The existing merge path preserves the same ownership filtering and resource construction as top-level nodes.TestRepositoryEnqueuesAgentSessionsOwnedByForeachBodyStepsverifies that removal enqueues the body session with its provider, ID, directory, worker, step name, and generation.
Full details: Out of Scope Changes check
-
Explanation
The changes are limited to foreach body session collection in
internal/persis/dagrun_repository.goand a focused repository test. The implementation supports the linked cleanup behavior and does not add unrelated product changes.
-
Full details: Docstring Coverage
-
Explanation
Docstring coverage is 50.00% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Documentation-only edits and deletions within surviving functions are included; unchanged diff context and fully deleted functions are excluded. Analyzed 2 functions across 2 files.
Resolution
Write docstrings for the functions missing them to satisfy the coverage threshold.
-
Finishing Touches
-
🧪 Generate unit tests (beta)
-
- Create a new PR
-
- Autofix · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.
Comment @coderabbitai help to get the list of available commands.
yohamta0
left a comment
yohamta0
left a comment
There was a problem hiding this comment.
Verdict: not safe to merge as is
The diagnosis and the fix direction are right — foreach body sessions genuinely never reach status.AgentSessions or any node, so run removal left them behind, and the regression test in dagrun_repository_test.go pins the behavior. Thanks for tracking this down from #3362 and for keeping the merge path shared with top-level steps.
But CI is red on this PR's own change, and the traversal strategy is heavier than it needs to be on the retention path.
Must fix before merging
- Lint failure (blocker).
Go Linterfails:internal/persis/dagrun_repository.go:423:17: G304: Potential file inclusion via variable (gosec). The repo convention is an annotated suppression — seeinternal/service/frontend/api/v1/dagruns_foreach.go:618(//nolint:gosec // paths derive from recorded log locations and validated segments) andinternal/persis/file/artifact/record.go:125. - Replace the recursive
WalkDirwith a deterministic read.WalkDirvisits every entry under the foreach step directory — all body-step stdout/stderr files — to find thestatus.jsonfiles. The layout is already addressable: readlogpath.ForeachItemsFilefor the item indexes, read<index>/status.json, and recurse only intologpath.ForeachNestedParent(item, step.Name)for body steps withForeach: true. This matters becauseenqueueDAGRunRemovalsalso runs for every expiring run inremoveOldDAGRuns(dagrun_repository.go:263), so a scheduler retention sweep multiplies the scan across runs. - Do not swallow read/unmarshal failures silently. A dropped item record means a provider session is leaked permanently — the run directory is deleted right after, so the record can never be reconstructed and there is no trace.
logger.Warnis already available in this file. - Don't alias the caller's slice.
result := nodesfollowed byappendwrites intostatus.Nodes' backing array whenevercap > len. Returning[]ir.AgentSessionResourceinstead of synthetic*ir.Nodevalues removes the aliasing and the memory retention noted below.
Worth discussing (not blocking)
- Layering.
internal/persisis the contracts/repository layer; file-backed I/O lives ininternal/persis/file. This change putsos.ReadFile+filepath.WalkDirin the repository and makes it the second place that encodes the foreach record layout (the frontend API is the first). A small reader next tointernal/cmn/logpathused by both would keep that knowledge in one place. - Previous attempts still leak.
agentSessionResourcesreads only the latest attempt, and each attempt gets a fresh log directory (logpath.Config.RunDirembeds a timestamp), so body-step sessions from an earlier attempt of a retried run are still never queued. Top-level steps avoid this becauseir.WithAgentSessionscarries their resources forward across attempts (internal/runtime/agent/agent.go:1588). The durable fix for both cases is to merge body-step sessions intostatus.AgentSessionswhen the foreach step finishes — removal then needs no change, retries carry over for free, and there is no delete-time disk scan. That needs a new way for an executor to report more than one owned session, so it is a larger change; your call whether to take it here or as a follow-up. - Test nit:
backend.eventsis wired but never asserted in the new test (onlyrecordingDAGRunRemovalEnqueuerrequires a non-nil pointer). Also, the test writes onlystatus.jsonand noitems.json, so it is coupled to theWalkDirimplementation — writing the records the way the runtime does would survive fix #2.
Performance / resource-safety assessment
Findings by inspection only; no tests, builds or profiling were run.
- Disk/I-O: the full-subtree
WalkDirper foreach node is the one real regression (item #2). Worst case for a foreach with many items and several body steps each producing stdout/stderr, multiplied by every run in a retention sweep. - Memory:
withForeachBodyNodesretainsstep.AgentSessionpointers for every item of every foreach node until the merge, and each session can carry up tomaxSessionEvents = 1000events plus interactions and permission grants. Converting toir.AgentSessionResourceat read time (or decoding into a trimmed struct) keeps peak memory flat in item count. - Cancellation: the walk never checks
ctx, so a sweep interrupted by shutdown keeps scanning. Returningctx.Err()from the walk function aborts it. - Queue growth: one cleanup record per owned body session is inherent to the fix and correct; worth knowing it can now be one record per foreach item.
- CPU, file count, goroutines/timers/FDs: none found by inspection. No new background work, and the walk closes each file via
os.ReadFile.
No prior review feedback existed on this PR (0 reviews, single commit 9d988fa5), so nothing to re-check against.
| if err != nil || d.IsDir() || d.Name() != logpath.ForeachItemStatusFile { | ||
| return nil | ||
| } | ||
| data, err := os.ReadFile(path) |
There was a problem hiding this comment.
CI blocker. Go Linter fails here: internal/persis/dagrun_repository.go:423:17: G304: Potential file inclusion via variable (gosec).
The repo convention is an annotated suppression at the call site, e.g. internal/service/frontend/api/v1/dagruns_foreach.go:618:
data, err := os.ReadFile(path) //nolint:gosec // paths derive from recorded log locations and validated segmentsIf the record reading moves behind a shared helper (see my comment on the function signature), the suppression lives there once instead.
There was a problem hiding this comment.
Fixed in e8dcd20. The read now goes through one readForeachRecord helper that carries the same //nolint:gosec annotation and reason as readJSONFile in the API.
| continue | ||
| } | ||
| stepDir := logpath.ForeachStepDir(node.Stdout, node.Step.Name) | ||
| _ = filepath.WalkDir(stepDir, func(path string, d fs.DirEntry, err error) error { |
There was a problem hiding this comment.
WalkDir scans the whole subtree to find status.json, so it also stats every body-step stdout/stderr file under the step directory. The layout is already addressable without a scan: read logpath.ForeachItemsFile in the step directory for the item indexes, read <index>/status.json for each, and recurse only into logpath.ForeachNestedParent(item, step.Name) for body steps whose record has Foreach: true. That reads exactly the records you need and matches how listForeachItems enumerates items.
This is not just a micro-optimization: enqueueDAGRunRemovals also runs for every expiring run in removeOldDAGRuns (line 263), so a scheduler retention sweep repeats this scan per run.
The walk also ignores cancellation — please take ctx and return ctx.Err() from the walk function (or check it per item) so an interrupted sweep stops promptly.
There was a problem hiding this comment.
Fixed in e8dcd20. The WalkDir is gone. The reader now follows the layout the runtime writes: items.json for the item list, then <item>/status.json per item, using logpath.ForeachParentDir, ForeachItemPath, ForeachItemDir and ForeachNestedParent. It also descends into nested foreach loops, and checks ctx.Err() per item.
| return nil | ||
| } | ||
| data, err := os.ReadFile(path) | ||
| if err != nil { |
There was a problem hiding this comment.
Swallowing the read error (and the json.Unmarshal error below) leaks a provider session permanently and silently: RemoveDAGRun deletes the run directory immediately after this, so the record is gone and the cleanup job can never be reconstructed. Every other failure path in agentSessionResources either propagates or is an explicit ErrNoStatusData/ErrDAGRunIDNotFound skip.
At minimum log it — logger is already imported in this file:
logger.Warn(ctx, "Failed to read foreach item record for agent session cleanup", tag.Error(err), tag.File(path))fs.ErrNotExist is the one case worth skipping quietly.
There was a problem hiding this comment.
Fixed in e8dcd20. Read and decode errors are now logged with the path. A missing file (fs.ErrNotExist) is skipped silently, since an item that never ran has no record. Only context cancellation aborts, so one damaged record can't block removing the run.
| // status; their sessions live in the item records under the foreach step's | ||
| // log directory, so cleanup has to read them from there. | ||
| func withForeachBodyNodes(nodes []*ir.Node) []*ir.Node { | ||
| result := nodes |
There was a problem hiding this comment.
result := nodes then append writes into the caller's backing array whenever cap(nodes) > len(nodes) — which a JSON-decoded status.Nodes can easily be. Nothing observes those slots today, but it mutates state owned by ReadStatus, which may hand out a shared status.
Returning []ir.AgentSessionResource instead of synthetic *ir.Node values fixes this and two other things: the caller can append the resources directly (no ir.Step{Name: ...} placeholder nodes), and you stop retaining a *ir.AgentSession per item until the merge — each one can carry up to maxSessionEvents = 1000 events plus interactions and permission grants, so a large foreach holds all of that live at once. Decoding into a trimmed record shape (step name + session identity fields) would avoid allocating the timelines at all.
There was a problem hiding this comment.
Fixed in e8dcd20. foreachBodySessionResources now returns []ir.AgentSessionResource built from its own slice, so it no longer appends onto the caller's nodes backing array.
| // body step that kept an agent session. Body steps are not nodes of the run | ||
| // status; their sessions live in the item records under the foreach step's | ||
| // log directory, so cleanup has to read them from there. | ||
| func withForeachBodyNodes(nodes []*ir.Node) []*ir.Node { |
There was a problem hiding this comment.
Layering: internal/persis holds the persistence contracts and repositories while file-backed I/O belongs in internal/persis/file, and this makes the repository the second place that encodes the foreach record layout — internal/service/frontend/api/v1/dagruns_foreach.go already has foreachStepDir, foreachItemRecord and readJSONFile doing the same reads.
Consider a small reader that owns "enumerate the item records of a foreach step" (next to internal/cmn/logpath, which already owns the path layout) and have both call sites use it. One place for the paths, one #nosec justification, one traversal to keep correct.
There was a problem hiding this comment.
Fixed in e8dcd20. The test now writes items.json and status.json the way the runtime does, including a nested foreach, and asserts both resources. It also checks that the enqueue event comes before the removal. I kept the shared reader inside persis for now rather than moving it next to logpath, to keep this PR small. I can follow up if you'd like the API to share it.
One limit to note: status.json only holds the latest record of an item, so a session from an earlier retried attempt of the same body step isn't recorded there and isn't cleaned up. That needs the runtime to keep discarded sessions, so I'd handle it separately.
Read items.json and each item's status.json directly instead of walking the step directory, descend into nested foreach loops, honour context cancellation, and log unreadable records instead of dropping them. Signed-off-by: Ryo Kusnadi <ryokusnadi@gmail.com>
|
Thanks for the review. I've addressed all five comments in e8dcd20 and replied on each thread. Could you take another look? |
Fixes #3362
Run cleanup built its list of provider sessions from the run status nodes only. A session owned by a foreach body step lives in the item records under the foreach step's log directory, so it was never queued for release.
agentSessionResourcesnow also reads the item records of each foreach node, including nested foreach steps, and merges their body steps' sessions into the same list. They are then released at the same time and in the same way as a top-level step's.I added
TestRepositoryEnqueuesAgentSessionsOwnedByForeachBodySteps, and theinternal/persistests andgo vetpass.Summary by CodeRabbit