From 2c196f20dfe572b8c569cc75917df6423422e793 Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Sat, 17 Oct 2026 20:51:14 +0800 Subject: [PATCH] Retire a conversation idle for retire_after storage.retention.retire_after is the second level of retention. A conversation idle that long has everything of it removed, its metadata too, and is retired: it is never collected again, even when its runtime resumes it. It is read on the same clock as metadata_only_after, the header time of the newest landed file, and is at least as long. It needs metadata_only_after set, and equal to it a conversation is retired with no metadata-only step first. Either rule broken is a configuration error. Only both settings off keeps everything. A pass retires at its end, after the send, before level 1, and gives no level 1 to an idle or a retired session. A session is retired once its chain covers its files, its metrics are derived when the pipeline derives them, and no removal of its files is in progress, asked under its locks. The steps keep a crash or a power loss from losing anything, or landing the session again: - The marker _retired/, an empty file, is written and synced, the errors checked, before anything is deleted. Every attempt that deletes anything writes it again. - Each metrics request the receiver does not have is written again from its receipt and synced, a request already there included. - The session's directory and its conversation's are renamed under _removed/ and deleted. Both directories of each rename are synced, the errors checked, before anything moved is deleted. - The requests the receiver has are deleted. The root, _conversations, _removed and the spool are synced on every attempt before push.state and metrics.state forget the session, and the provider's table marks its lines retired. A session's owner stays in push.state while a request of it waits in the spool, so the request goes to that owner only. Every pass finishes a retirement left unfinished before anything lands, whatever the settings say then, and asz parse and asz push do the same first. The pass line counts retired=, the startup says what each level does, and the page drops a retired conversation from memory, also one another process retired. --- CLAUDE.md | 23 +- asz.yaml | 8 +- cmd/asz/main.go | 6 +- cmd/asz/parse.go | 14 +- cmd/asz/refresh.go | 25 +- cmd/asz/retention_test.go | 227 +++- cmd/asz/view.go | 14 +- docs/en/adapters/claude-code.md | 5 +- docs/en/adapters/kiro.md | 4 + docs/en/adapters/langsmith.md | 12 + docs/en/changes/changes.md | 26 + docs/en/formats/storage-root.md | 129 +- docs/en/guides/command-line-internals.md | 24 +- docs/en/guides/scenario.md | 13 +- docs/en/setup/configuration.md | 3 +- internal/adapters/claudecodeprovider/seen.go | 32 + internal/config/config.go | 41 +- internal/config/config_test.go | 62 + internal/export/otlp/sent.go | 33 +- internal/metrics/metrics.go | 2 +- internal/metrics/receipt.go | 72 ++ internal/retention/retention.go | 83 +- internal/retention/retire.go | 474 +++++++ internal/storage/removed.go | 33 +- internal/view/cache_test.go | 54 + internal/view/view.go | 28 +- tests/chain/retire_test.go | 1190 ++++++++++++++++++ tests/chain/retired_test.go | 7 + tests/langchain_retire_test.go | 92 ++ 29 files changed, 2647 insertions(+), 89 deletions(-) create mode 100644 internal/retention/retire.go create mode 100644 tests/chain/retire_test.go create mode 100644 tests/langchain_retire_test.go diff --git a/CLAUDE.md b/CLAUDE.md index dfbf17b..5714d1a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -69,7 +69,7 @@ internal/view/conversation-view/ Horizon's conversation renderer, built from the internal/export/otlp/ sends landed files and rounds to an OpenTelemetry logs receiver, one record per file, and the metrics spool internal/metrics/ derives the runtime's token metric from landed Session Data into the spool; collector side; testdata holds a redacted capture of the runtime's exporter internal/repack/ re-cuts landed files into a new root under another budget -internal/retention/ keeps only idle conversations' metadata: removes landed files, a skeleton in each one's place (collector side) +internal/retention/ the two levels for idle conversations: keeps only the metadata, a skeleton in each landed file's place, then retires the session whole (collector side) internal/adapters/claudecode/ the claude-code-local adapter internal/adapters/claudecodeprovider/ the claude-code-provider adapter: the request and response bodies Claude Code writes, landed into their sessions internal/adapters/kiro/ the kiro-local adapter: the sessions Kiro's V3 engine writes, for kiro-cli and Kiro Crew @@ -180,12 +180,21 @@ Every source file carries the Apache-2.0 header; `make license-fix` inserts miss and a reader may have taken the file) and is reported once per directory; failing it made a disk that kept failing land the same records on every pass. Retention, when it removes, must make a replacement durable, its directory synced and the error checked, before it deletes what it replaces. -- **asz never removes a real session whole.** Only a session a scenario build marked is removed - whole, by the policy in its marker, and only once all of it was sent to the one receiver the - pipeline sends to. A removal renames each directory away before it deletes it, so no reader sees - half a session. `storage.retention.metadata_only_after`, off by default, removes the landed - files of an idle conversation, sent or not, and keeps its structure: a skeleton in each file's - place, and the index, cursors, state and chain untouched. It never runs over a scenario root. +- **Only `retire_after` removes a real session whole, and its id is never collected again.** A + scenario build's removal takes only a session the build marked, by the policy in its marker, and + only once all of it was sent to the one receiver the pipeline sends to. A removal renames each + directory away before it deletes it, so no reader sees half a session. `storage.retention` has + two levels on one clock, the header time of the newest landed file, sent or not: off by default, + `metadata_only_after` removes the landed files of an idle conversation and keeps its structure, + a skeleton in each file's place and the index, cursors, state and chain untouched; + `retire_after`, at least as long, removes everything and retires the session. Neither runs over + a scenario root. +- **A retired session is never written again.** Its marker `_retired/` is an empty file, + written and synced before anything of the session is deleted, and never removed. Every writer + asks `storage.Retired` before the first `MkdirAll` or lock that would make the session's or its + chain's directory, since taking a lock makes it; a marker that cannot be read is an error, never + taken for none. A retired session's unsent metrics requests are written again from their + receipts before its directory goes, and keep its owner in `push.state` until they are sent. - **A landed file is cut once.** A round addresses records by `{seq, row}` and binds to file digests, so a referenced file is never re-cut. A new budget applies to new files; an existing root is brought under it with `asz repack` into a new root, where the chains are built again. diff --git a/asz.yaml b/asz.yaml index b2842d9..fd744d2 100644 --- a/asz.yaml +++ b/asz.yaml @@ -33,7 +33,7 @@ storage: # How long a conversation may be idle, with nothing new landed in any of # its streams, before this root keeps less of it. Whether it was sent, and # whether an endpoint is set, does not matter. Each value is a duration - # such as 72h, or days such as 30d, and -1 is off. Only off keeps + # such as 72h, or days such as 30d, and -1 is off. Only both off keeps # everything. retention: # To save disk, keep only the conversation's metadata: what was said and @@ -42,6 +42,12 @@ storage: # conversation. The removed content cannot be restored here; only a # receiver that accepted it still has it. metadata_only_after: -1 + # Remove everything of the conversation, its metadata too, and retire + # it. A retired conversation is never collected again, even when its + # runtime resumes it, so set this longer than the runtime keeps its own + # history. It is at least metadata_only_after, which must be set too; + # equal to it goes straight to retirement. + retire_after: -1 adapters: - name: claude-code-local diff --git a/cmd/asz/main.go b/cmd/asz/main.go index 237f384..b65c378 100644 --- a/cmd/asz/main.go +++ b/cmd/asz/main.go @@ -639,7 +639,7 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error { if retired, err := storage.RetiredSessions(storage.NewZone(zoneRoot)); err != nil { return err } else if len(retired) > 0 { - fmt.Printf("%d session(s) are retired: nothing of them is kept, and none is collected again\n", len(retired)) + fmt.Printf("%d session(s) are retired, and none is collected again\n", len(retired)) } // A conversation's rounds are checked too. They are a digest chain, so a @@ -861,11 +861,11 @@ func cmdCollect(cfg *config.Config, ads []config.Adapter, once bool) error { } defer closeClient() ref.pusher = pusher - keep, err := cfg.Storage.Retention.MetadataOnlyAfterDuration() + metadataOnly, retire, err := cfg.Storage.Retention.Durations() if err != nil { return err } - ref.retain(keep) + ref.retain(metadataOnly, retire) if once || ref.base.Mode == config.ModeOnce { // Nothing follows a single pass, so whatever it could not do is // reported rather than left for a pass that never comes. diff --git a/cmd/asz/parse.go b/cmd/asz/parse.go index 19eb104..24da975 100644 --- a/cmd/asz/parse.go +++ b/cmd/asz/parse.go @@ -29,6 +29,7 @@ import ( "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode" "github.com/apache/skywalking-ai-sessionizer/internal/config" "github.com/apache/skywalking-ai-sessionizer/internal/parse" + "github.com/apache/skywalking-ai-sessionizer/internal/retention" "github.com/apache/skywalking-ai-sessionizer/internal/storage" "github.com/apache/skywalking-ai-sessionizer/internal/view" "github.com/apache/skywalking-ai-sessionizer/pkg/model" @@ -64,6 +65,10 @@ var rootWait = exportWait // waits for a pass of another command that holds it. The root has to be // there first: taking the lock makes a directory under it, which would make // a mistyped path look like an empty root. +// +// It then finishes every retirement a pass left unfinished, as a pass does +// first. A retirement that cannot finish is said and the command goes on: +// its marker is there, so the command writes nothing of that session. func holdRootFor(root, command string) (*storage.SessionLock, error) { if _, err := os.Stat(root); err != nil { return nil, err @@ -72,7 +77,14 @@ func holdRootFor(root, command string) (*storage.SessionLock, error) { if errors.Is(err, storage.ErrRootBusy) { return nil, fmt.Errorf("%s: another command held this storage root for %s; nothing was done, run it again", command, rootWait) } - return lock, err + if err != nil { + return nil, err + } + res := (&retention.Remover{Zone: storage.NewZone(root)}).RecoverRetired() + for _, e := range res.Errors { + fmt.Fprintf(os.Stderr, " error: %s: %v\n", command, e) + } + return lock, nil } // parseZone assembles every session under zoneRoot, or the one named by want, diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go index 4cea624..7ffd0fb 100644 --- a/cmd/asz/refresh.go +++ b/cmd/asz/refresh.go @@ -438,14 +438,17 @@ func present(dir string, optional bool) (bool, error) { } // retain sets the retention: how long a conversation may be idle before only -// its metadata is kept. Zero removes nothing, though a removal an earlier -// pass left in progress is still finished. -func (r *refresher) retain(keep time.Duration) { - r.retention = &retention.Remover{Zone: r.zone, MetadataOnlyAfter: keep, Metrics: r.deriver != nil} +// its metadata is kept, and before it is retired. Zero does nothing at that +// level, though a removal or a retirement an earlier pass left in progress is +// still finished. +func (r *refresher) retain(metadataOnly, retire time.Duration) { + r.retention = &retention.Remover{Zone: r.zone, MetadataOnlyAfter: metadataOnly, RetireAfter: retire, Metrics: r.deriver != nil} } // forget drops what the page cached of sessions whose files were removed, -// so it reads their skeletons from then on. +// so it reads their skeletons from then on, or that were retired: the list +// no longer names a retired conversation, so nothing would read it again to +// drop it. func (r *refresher) forget(sessions []string) { if r.srv == nil { return @@ -526,8 +529,11 @@ func (r *refresher) pass() error { // dropped counts the LangSmith runs not collected this pass because their // conversation could not be told. var dropped int - // retired names the retired sessions a collector found evidence of. + // retired names the retired sessions a collector found evidence of, and + // retiredNow counts the sessions retired this pass, by a retirement that + // began in it or one it finished. var retired []string + retiredNow := 0 changed := map[string]bool{} // A removal an earlier pass left in progress is finished first, before @@ -536,6 +542,8 @@ func (r *refresher) pass() error { res := r.retention.Recover() errs = append(errs, res.Errors...) r.forget(res.Removed) + r.forget(res.Retired) + retiredNow = len(res.Retired) } var cs *claudecode.Stats @@ -843,7 +851,9 @@ func (r *refresher) pass() error { res := r.retention.Pass(retention.Input{Busy: r.removeInput().Busy}) errs = append(errs, res.Errors...) retained = res.Files + retiredNow += len(res.Retired) r.forget(res.Removed) + r.forget(res.Retired) } if r.last && len(r.retry) > 0 { // No pass follows, so these are not deferred, they are unparsed. @@ -879,6 +889,9 @@ func (r *refresher) pass() error { if retained > 0 { pushed += fmt.Sprintf(" retention=%d", retained) } + if retiredNow > 0 { + pushed += fmt.Sprintf(" retired=%d", retiredNow) + } bodies := waiting != r.waiting || unreadable != r.unreadable r.waiting, r.unreadable = waiting, unreadable if first || landed > 0 || rounds > 0 || busy > 0 || pushed != "" || len(errs) > 0 || bodies || dropped > 0 { diff --git a/cmd/asz/retention_test.go b/cmd/asz/retention_test.go index 7547a7d..960a9a6 100644 --- a/cmd/asz/retention_test.go +++ b/cmd/asz/retention_test.go @@ -18,6 +18,7 @@ package main import ( + "errors" "os" "path/filepath" "strings" @@ -27,6 +28,8 @@ import ( "github.com/apache/skywalking-ai-sessionizer/internal/adapters/kiro" "github.com/apache/skywalking-ai-sessionizer/internal/config" "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp" + "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest" + "github.com/apache/skywalking-ai-sessionizer/internal/retention" "github.com/apache/skywalking-ai-sessionizer/internal/scenario" "github.com/apache/skywalking-ai-sessionizer/internal/storage" "github.com/apache/skywalking-ai-sessionizer/internal/view" @@ -45,7 +48,7 @@ func removedRoot(t *testing.T) (string, string, *refresher) { rcv := receiver(t) ref := pipelineOver(t, root, bothAdapters(root, "")) sendingTo(t, ref, rcv) - ref.retain(time.Nanosecond) + ref.retain(time.Nanosecond, 0) out := stdoutOf(t, func() { if err := ref.pass(); err != nil { t.Fatal(err) @@ -109,7 +112,7 @@ func TestRetentionLeavesAScenarioRootAlone(t *testing.T) { // the session, and the pass sends it. ref := pipelineOver(t, root, bothAdapters(root, t.TempDir())) sendingTo(t, ref, rcv) - ref.retain(time.Nanosecond) + ref.retain(time.Nanosecond, 0) stderrOf(t, func() { if err := ref.pass(); err != nil { t.Fatal(err) @@ -237,7 +240,7 @@ func unsentRemovedRoot(t *testing.T, scenarioName string) (string, string) { t.Fatal(err) } t.Cleanup(ref.close) - ref.retain(time.Nanosecond) + ref.retain(time.Nanosecond, 0) stdoutOf(t, func() { if err := ref.pass(); err != nil { t.Fatal(err) @@ -422,7 +425,7 @@ func TestCommandsSayWhatIsRetired(t *testing.T) { t.Fatalf("verify: %v", err) } }) - if !strings.Contains(out, "1 session(s) are retired: nothing of them is kept, and none is collected again") { + if !strings.Contains(out, "1 session(s) are retired, and none is collected again") { t.Errorf("verify does not say the session is retired:\n%s", out) } @@ -505,3 +508,219 @@ func TestEverySourcesListNamesARetiredSession(t *testing.T) { }) } } + +// A pass retires a conversation idle for retire_after, after it sends, and +// its line counts it. The page stops listing it. The next pass retires +// nothing more, and its source, which is still there, is not collected +// again. +func TestAPassRetiresAnIdleConversation(t *testing.T) { + root, b := scenarioRoot(t, "growth.yaml") + if err := os.RemoveAll(filepath.Join(root, storage.ScenarioDir)); err != nil { + t.Fatal(err) + } + rcv := receiver(t) + ref := pipelineOver(t, root, bothAdapters(root, "")) + sendingTo(t, ref, rcv) + ref.retain(time.Nanosecond, time.Nanosecond) + out := stdoutOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + if !strings.Contains(out, " retired=1") { + t.Fatalf("the pass line does not count the retirement:\n%s", out) + } + zone := storage.NewZone(root) + if retired, err := storage.Retired(zone, b.Session); err != nil || !retired { + t.Fatalf("no marker: %v", err) + } + if pathExists(zone.SessionDir(b.Session)) || pathExists(filepath.Join(root, "_conversations", b.Session)) { + t.Fatal("a directory of the retired session is left") + } + if ids, err := ref.srv.List(); err != nil || len(ids) != 0 { + t.Fatalf("the page lists %v, %v", ids, err) + } + stderrOf(t, func() { + out = stdoutOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + }) + if strings.Contains(out, " retired=") { + t.Fatalf("the next pass retired something:\n%s", out) + } + if pathExists(zone.SessionDir(b.Session)) { + t.Fatal("the next pass collected the retired session again") + } +} + +// A retired conversation may never be read again. The pass must drop it +// from the page's memory, whether retirement finishes in the pass or in +// recovery. Only the pass that finishes the retirement counts it. +func TestAPassForgetsARetiredConversation(t *testing.T) { + for _, finish := range []string{"pass", "recovery"} { + t.Run(finish, func(t *testing.T) { + root, b := scenarioRoot(t, "growth.yaml") + if err := os.RemoveAll(filepath.Join(root, storage.ScenarioDir)); err != nil { + t.Fatal(err) + } + rcv := receiver(t) + ref := pipelineOver(t, root, bothAdapters(root, "")) + sendingTo(t, ref, rcv) + stdoutOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + if _, err := ref.srv.Load(b.Session); err != nil { + t.Fatal(err) + } + if !ref.srv.Kept(b.Session) { + t.Fatal("the page did not keep the conversation it loaded") + } + + zone := storage.NewZone(root) + chain := filepath.Join(root, "_conversations", b.Session) + ref.retain(time.Nanosecond, time.Nanosecond) + if finish == "recovery" { + stopped := errors.New("stopped as a crash would stop it") + ref.retention.Stop = func(p retention.Point) error { + if p.Phase == retention.PhaseRetireMarked { + return stopped + } + return nil + } + var err error + var out string + stderrOf(t, func() { + out = stdoutOf(t, func() { err = ref.pass() }) + }) + if !errors.Is(err, stopped) { + t.Fatalf("the retirement did not stop at its marker: %v", err) + } + if !strings.Contains(out, "] refreshed:") || strings.Contains(out, " retired=") { + t.Fatalf("the stopped pass should print its line without counting a retirement:\n%s", out) + } + if !pathExists(zone.RetiredPath(b.Session)) || !pathExists(zone.SessionDir(b.Session)) || !pathExists(chain) { + t.Fatal("the stopped retirement should leave its marker, session and chain") + } + if _, err := ref.srv.Load(b.Session); err != nil { + t.Fatal(err) + } + if !ref.srv.Kept(b.Session) { + t.Fatal("the page did not keep the conversation before recovery") + } + ref.retention.Stop = nil + } + + var out string + stderrOf(t, func() { + out = stdoutOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + }) + if ref.srv.Kept(b.Session) { + t.Error("the page still keeps the retired conversation without another read") + } + if !strings.Contains(out, "] refreshed:") || !strings.Contains(out, " retired=1 ") { + t.Errorf("the pass line does not count the finished retirement:\n%s", out) + } + if !pathExists(zone.RetiredPath(b.Session)) || pathExists(zone.SessionDir(b.Session)) || pathExists(chain) { + t.Error("the retirement should leave its marker and remove the session and chain") + } + }) + } +} + +// The pipeline says at startup what each level does. Equal levels go +// straight to retirement, so level 1 is not said then. +func TestThePipelineSaysWhatEachLevelDoes(t *testing.T) { + root, _ := scenarioRoot(t, "growth.yaml") + ref := pipelineOver(t, root, bothAdapters(root, "")) + cfg := config.Default() + const metadataOnly, retired = "keeps only its metadata", "is retired: everything of it is removed, and it is never collected again" + for _, c := range []struct { + metadataOnly, retire time.Duration + says, not []string + }{ + {time.Hour, 0, []string{"idle for 1h0m0s " + metadataOnly}, []string{retired}}, + {time.Hour, 2 * time.Hour, []string{"idle for 1h0m0s " + metadataOnly, "idle for 2h0m0s " + retired}, nil}, + {2 * time.Hour, 2 * time.Hour, []string{"idle for 2h0m0s " + retired}, []string{metadataOnly}}, + {0, 0, nil, []string{metadataOnly, retired}}, + } { + ref.retain(c.metadataOnly, c.retire) + out := stdoutOf(t, func() { printPipeline(ref, cfg) }) + for _, want := range c.says { + if !strings.Contains(out, want) { + t.Errorf("%v and %v: the startup does not say %q:\n%s", c.metadataOnly, c.retire, want, out) + } + } + for _, not := range c.not { + if strings.Contains(out, not) { + t.Errorf("%v and %v: the startup says %q:\n%s", c.metadataOnly, c.retire, not, out) + } + } + } +} + +// asz parse and asz push finish a retirement a pass left unfinished before +// they do anything else, whatever the configuration says now, and write +// nothing of the session: no round, no index, nothing sent. +func TestStandaloneCommandsFinishARetirementFirst(t *testing.T) { + for _, command := range []string{"parse", "push"} { + t.Run(command, func(t *testing.T) { + root, b := scenarioRoot(t, "growth.yaml") + if err := os.RemoveAll(filepath.Join(root, storage.ScenarioDir)); err != nil { + t.Fatal(err) + } + rcv := receiver(t) + ref := pipelineOver(t, root, bothAdapters(root, "")) + sendingTo(t, ref, rcv) + ref.retain(time.Nanosecond, time.Nanosecond) + ref.retention.Stop = func(p retention.Point) error { + if p.Phase == retention.PhaseRetireMarked { + return errors.New("stopped as a crash would stop it") + } + return nil + } + stderrOf(t, func() { + stdoutOf(t, func() { _ = ref.pass() }) + }) + zone := storage.NewZone(root) + if !pathExists(zone.SessionDir(b.Session)) || !pathExists(zone.RetiredPath(b.Session)) { + t.Fatal("the stopped retirement should leave its marker and the session's directory") + } + rcv.Reset() + cfg := &config.Config{Metrics: allMetrics} + cfg.Storage.Root = root + o := rcv.Options(otlp.ProtocolGRPC) + cfg.Export.OTLP.Protocol, cfg.Export.OTLP.Endpoint, cfg.Export.OTLP.ServiceName = o.Protocol, o.Endpoint, "Removal Test" + var err error + stderrOf(t, func() { + stdoutOf(t, func() { + if command == "parse" { + err = cmdParse(cfg, config.Adapter{}, false) + } else { + err = cmdPush(cfg, config.Adapter{}, false) + } + }) + }) + if err != nil { + t.Fatal(err) + } + for _, dir := range []string{zone.SessionDir(b.Session), filepath.Join(root, "_conversations", b.Session)} { + if pathExists(dir) { + t.Fatalf("%s is left after asz %s", dir, command) + } + } + for _, rec := range otlptest.Records(rcv.Requests()) { + if otlptest.Attrs(rec.Attributes)["asz.session"] == b.Session { + t.Fatalf("asz %s sent a file of the retired session", command) + } + } + }) + } +} diff --git a/cmd/asz/view.go b/cmd/asz/view.go index bd7ecf3..6bbda7d 100644 --- a/cmd/asz/view.go +++ b/cmd/asz/view.go @@ -166,11 +166,11 @@ func cmdServer(cfg *config.Config, ads []config.Adapter, once bool) error { } defer closeClient() ref.pusher = pusher - keep, err := cfg.Storage.Retention.MetadataOnlyAfterDuration() + metadataOnly, retire, err := cfg.Storage.Retention.Durations() if err != nil { return err } - ref.retain(keep) + ref.retain(metadataOnly, retire) if once || ref.base.Mode == config.ModeOnce { // One pass and no loop after it, exactly as asz collect -once has: // what another process was holding is reported, not deferred. @@ -262,8 +262,14 @@ func printPipeline(ref *refresher, cfg *config.Config) { } fmt.Printf("export : %s, at the end of every pass\n", endpoint) } - if ref.retention != nil && ref.retention.MetadataOnlyAfter > 0 { - fmt.Printf("retention: a conversation idle for %s keeps only its metadata; its content is removed, and only a receiver that accepted it still has it\n", ref.retention.MetadataOnlyAfter) + if rt := ref.retention; rt != nil { + // Equal levels go straight to retirement, so level 1 does nothing. + if rt.MetadataOnlyAfter > 0 && (rt.RetireAfter == 0 || rt.MetadataOnlyAfter < rt.RetireAfter) { + fmt.Printf("retention: a conversation idle for %s keeps only its metadata; its content is removed, and only a receiver that accepted it still has it\n", rt.MetadataOnlyAfter) + } + if rt.RetireAfter > 0 { + fmt.Printf("retention: a conversation idle for %s is retired: everything of it is removed, and it is never collected again, even when its runtime resumes it\n", rt.RetireAfter) + } } // A root a claude-code scenario build wrote. Only there does a pipeline // remove a session whole, and only one a build marked. diff --git a/docs/en/adapters/claude-code.md b/docs/en/adapters/claude-code.md index 245b322..1efdceb 100644 --- a/docs/en/adapters/claude-code.md +++ b/docs/en/adapters/claude-code.md @@ -124,7 +124,10 @@ that is not proven. A cursor's size check and its tail digest catch it, and the that source with a conflict rather than read from a position that no longer means what it did. A session is never finished. A transcript quiet for weeks can grow again when the session is -resumed, so the adapter never marks a session done, and every pass checks every source again. +resumed, so the adapter never marks a session done, and every pass checks every source again. A +session retired by `storage.retention.retire_after` is the one exception: it is never collected +again, even when Claude Code resumes it. Set `retire_after` longer than `cleanupPeriodDays`. +See [Retirement](../formats/storage-root.md#retirement). Collection polls, as [collector](../setup/configuration.md#collector) describes. There is no file system watcher. diff --git a/docs/en/adapters/kiro.md b/docs/en/adapters/kiro.md index 64d8bff..2f4ddc9 100644 --- a/docs/en/adapters/kiro.md +++ b/docs/en/adapters/kiro.md @@ -35,6 +35,10 @@ in `asz sources` and once in `asz collect`: not read : kiro-local: 24 session(s) of kiro-cli's V2 engine under sessions/cli; only the V3 engine's sessions are collected: start kiro-cli with --v3, or set chat.agentEngine to v3 ``` +Kiro keeps a session until a person or Kiro Crew deletes it. A session retired by +`storage.retention.retire_after` is never collected again, its sub-executions included, even when +Kiro resumes it. See [Retirement](../formats/storage-root.md#retirement). + ## What is read One session directory holds: diff --git a/docs/en/adapters/langsmith.md b/docs/en/adapters/langsmith.md index 998c37f..bd1cdbf 100644 --- a/docs/en/adapters/langsmith.md +++ b/docs/en/adapters/langsmith.md @@ -352,6 +352,18 @@ start seen, it lands under a name that says the identity was not supplied, and n with anything else. A later update of one of its runs that carries a key stays there: one run is one conversation. +The client keeps nothing, so asz alone gives a conversation its lifetime, by the same two levels of +`storage.retention` as every runtime. A conversation retired by `retire_after` is never collected +again. A later run of its thread is not collected, and neither is an update of one of its runs, or a +run inside one that names no conversation of its own, while asz remembers that run. They do not wait +for anything. What `open.json` remembers of each run is what tells them apart, and a retirement does +not change it. A run that names a conversation of its own is checked directly. The same holds for +an update with a complete identity, even if asz no longer remembers its run. It is not collected +if that conversation is retired. Otherwise, it lands in that conversation. An update of a run asz +no longer remembers waits as long as it may only when its own identity is incomplete. If that +identity remains unresolved, the update is then not collected. See +[Retirement](../formats/storage-root.md#retirement). + ## What the wire cannot supply - **A child's own conversation.** An agent invoked with its own `thread_id` reports the parent's on diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 9db0674..1f0b97d 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -346,6 +346,32 @@ what an answer left out. The renderer is Horizon's, now pinned at apache/skywalking-horizon-ui#192, whose prompt store holds skeletons, takes a piece by the SHA-256 a removed or too large part keeps, from the record that landed first, and reads kept bytes. +- A conversation can be retired. `storage.retention.retire_after`, `-1` and off by default, is how + long a conversation may be idle before everything of it is removed, its metadata too, and it is + retired. A retired conversation is never collected again, even when its runtime resumes it. Set + `retire_after` longer than the runtime keeps its own history. Idle is read on the same clock + as `metadata_only_after`, which must be set too, and `retire_after` is at least as long; equal, + a conversation is retired with no metadata-only step first. Either rule broken is a + configuration error. Only both settings off keeps everything. `asz collect` and `asz server` + retire at the end of a pass, after the send, a conversation whose head round covers its files + and, with metrics derived, every file of which is derived. See + [Retirement](../formats/storage-root.md#retirement). +- A retirement writes its marker, `_retired/`, an empty file, and syncs it before it + deletes anything. Next, it writes again, from its receipt, each metrics request the receiver does + not have, and syncs it. It then moves the session's directory and its conversation's under + `_removed/` and deletes them there. It syncs every directory it changed before it takes the + session out of `push.state` and `metrics.state`. The provider's table keeps its lines, marked + `retired`. Requests not sent stay until they are sent, to the session's owner only, whose line + stays meanwhile. A crash or a failed sync at any point is + finished by the next pass before anything lands, whatever the settings say then, and `asz parse` + and `asz push` finish it first too. The pass line counts `retired=`, and the startup says what + each level does. +- Nothing of a retired session is written again. Every collector passes it over before it makes a + directory, the provider's bodies of it included, and a LangSmith run of it never waits. The + metrics deriver and the parser do nothing for it, `asz push` sends none of its files or rounds + but still sends its waiting metrics requests, and `asz repack`, an sd scenario build and the + scenario runner refuse it. The pass log says once a process that a retired session's source was + found and not collected, `asz sources` names it, and `asz verify` counts the retired sessions. ## Configuration diff --git a/docs/en/formats/storage-root.md b/docs/en/formats/storage-root.md index 1f774f9..8aff2eb 100644 --- a/docs/en/formats/storage-root.md +++ b/docs/en/formats/storage-root.md @@ -12,6 +12,7 @@ session. .lock one collector per session removing.json a removal in progress, see Retention removal.state the generation of the last removal that deleted files + langsmith.shape.json where LangSmith runs were placed, when it collects them provider/ the bytes of removed chain tips and pieces, see Retention metrics/ immutable decisions of local metric derivation 000001.json exact request and accounting for landed file 1 @@ -62,17 +63,25 @@ session. .lock one pusher per root _scenario/ only in a root a claude-code scenario build wrote .lock one pipeline per scenario root - _removed/ what a removal of a scenario session is deleting - / the session directory, moved here whole - .chain/ its conversation's directory, moved here whole + _langsmith/ when the LangSmith receiver runs + runs--.request a request received and not landed yet + unreadable/ requests set aside, each with why + open.json the runs remembered, and the conversation of each + .lock one LangSmith collector per root + _retired/ the retired sessions + an empty file: the session is never collected again + _removed/ what a removal of a scenario session, or a retirement, + / is deleting: the session directory, moved here whole, + .chain/ and its conversation's directory, moved here whole ``` Child streams are flat siblings of `main`, keyed by agent id. Provider bodies sit under the session, not under a stream, because a body names no stream; the view joins each to its call. The storage path deliberately does not mirror the source tree: a path must not encode a relationship the pipeline is supposed to derive. [OTLP Records](otlp.md) says how the metrics spool fills and -what `push.state` records. `_scenario/` and `_removed/` exist only in a root a scenario build -wrote. [Retention](#retention) says what they are for. +what `push.state` records. `_scenario/` exists only in a root a scenario build wrote, and +`_retired/` once a session is retired. `_removed/` is used by a scenario's removal and by a +retirement. [Retention](#retention) says what each is for. Every command that writes to a root holds `_root/.lock` for one pass: `asz collect` and `asz server` for each pass, `asz parse`, `asz push` and `asz repack` for their run, and `asz scenario @@ -314,7 +323,10 @@ and a session of thousands of calls would need thousands of snapshot cursors. So one table for the root, `_provider/seen.state`: for each body file, its size, its modification time, the start of its digest, its state, its session, and the ids that join it. The states are `landed`, `waiting` for a session, `excluded` by the filter, `unreadable`, `changed` after it landed, -and `conflict` with a body of the same name the session already holds. +`conflict` with a body of the same name the session already holds, and `retired`: a body of a +retired session, which never lands again. Claude Code never deletes a body file, so a `retired` +line is what keeps each pass from reading the file again. It goes once its file is gone. A body +whose request names a retired session, or answers a `retired` request, becomes `retired` too. The table is derived. Every landed body is a record whose manifest names its file, its size and its digest, so a missing table is rebuilt from the landed records, and the first pass after reads each @@ -391,7 +403,8 @@ another machine with no source files and no collector. Session Data is already c records name identifiers by role, not by any runtime's fields. So the index rebuilds from the landed files with no adapter and no runtime-specific code, and the chain head is recovered from the rounds directory. `asz verify` checks every landed file against its digest and every round against -its commit digest without touching the source. +its commit digest without touching the source. A copy that collects again takes `_retired/` with +it, or it collects a retired session again from its start. For the same reason, the index, assembly and the chain hold no code for any one runtime. Supporting another runtime needs an adapter that writes Session Data, and nothing on the assembling side. The @@ -418,13 +431,20 @@ The store had also outlived its source: it held 48 sessions where the source dir ### storage.retention -`storage.retention` says how long an idle conversation keeps what in the root. A conversation is -idle for as long as nothing new has landed in any of its streams, by asz's clock: the time is read -from the header of its newest landed file. Whether it was sent, and whether an endpoint is set, does -not change that. Each setting is `-1`, off, by default, and only off keeps everything. +`storage.retention` says how long an idle conversation keeps what in the root, at two levels on one +clock. A conversation is idle for as long as nothing new has landed in any of its streams, by asz's +clock: the time is read from the header of its newest landed file or skeleton, never from a +record's time, a file's time or when a file was removed. Whether it was sent, whether an endpoint +is set, and whether its source is still there do not change that. Each setting is `-1`, off, by +default, and only both off keeps everything. + +- `metadata_only_after` keeps only its metadata, to save disk, as this section describes. +- `retire_after` removes everything of it and retires it, as [Retirement](#retirement) describes. + It is at least `metadata_only_after`, which must be set too, or the configuration is refused. `storage.retention.metadata_only_after` keeps only the metadata of a conversation idle for longer -than it, to save disk: it removes the landed files and keeps a skeleton of each. A pipeline, `asz +than it, to save disk: it removes the landed files and keeps a skeleton of each. A conversation idle +for `retire_after` is retired instead, so with both the same, no skeleton is written. A pipeline, `asz collect` or `asz server`, removes at the end of a pass, after the send. A session's files go together, every landed file of it through its newest, never some of them. A session keeps them while any of these holds: @@ -533,6 +553,80 @@ reads them again, and after four reads that each raced one it reports the check rather than as damage. `asz repack` refuses a session with skeletons or with a removal in progress: a skeleton keeps no records to re-cut. +### Retirement + +`storage.retention.retire_after` retires a conversation idle for longer than it: everything of it is +removed, its metadata too, and a marker keeps it from ever being collected again. That holds even +when its runtime resumes it, so set `retire_after` longer than the runtime keeps its own history. +Claude Code keeps a transcript for `cleanupPeriodDays`, 30 days when it is not set, and a session +whose transcript it removed cannot be resumed. Kiro keeps its sessions until a person or Kiro Crew +deletes them. A LangChain conversation has no source of its own: asz alone gives it a lifetime. + +A pipeline, `asz collect` or `asz server`, retires at the end of a pass, after the send, before it +looks at level 1. A conversation is retired when all of these hold: + +- Nothing has landed for it for `retire_after`. Its files may be landed files, skeletons or both. +- No removal of its files is in progress. +- Its head round reaches its newest landed file, and its parse did not fail this pass. +- With metrics derived, every landed file of it is derived, so each file's request exists. + +Sending is not waited for. A conversation idle for `retire_after` is never given level 1, even +while it waits to be retired. + +A retirement is written so that a crash or a power loss at any point loses nothing, and lands +nothing of the session again: + +1. `_retired/`, an empty file, is written, and its directory and the root are synced, + the errors checked. A failed sync stops the retirement there. The next attempt writes the marker + again, since a later sync that succeeds does not say an earlier one's data reached the disk. From + here on, every writer passes the session over. +2. Every metrics request of the session the receiver has not answered is written again from its + receipt, and synced, the error checked. A request already there is written again too: its sync + may have failed. It is read-only, and on Windows a rename cannot replace a read-only file, so it + is deleted first. The receipt is still there, so a crash between the two leaves what writes it + again. +3. The session's directory is renamed to `_removed//`, and its conversation's to + `_removed/.chain/`, and each is deleted there. The lock inside each is released + first, since Windows cannot rename a directory that holds an open lock. After each rename, the + two directories it changed are synced, the errors checked, before anything moved is deleted. +4. Every metrics request of the session the receiver has, a partly rejected one included, is + deleted. Then the root, `_conversations/`, `_removed/` and the spool are synced, the errors + checked. Every attempt syncs them, so what an earlier attempt changed is durable too: a rename + or a deletion cannot be made again, and a later sync of its directory is taken for it, as for + the deletions of level 1. A power + loss that brought back the session's directory after the state forgot what the receiver has + would make those requests again from their receipts, and send them. +5. `push.state` loses every line of the session whose file is gone. Its owner line stays while a + request of it waits in the spool: a request goes to its session's owner only, and with no owner + it would go to whichever owner the configuration names then. `metrics.state` loses the session, + and the provider's table marks its lines `retired`. `_langsmith/open.json` is not changed: what + it remembers of a run is what tells a later update of it to be retired, and it forgets a run by + age, as before. + +The marker goes first because, after it, a crash anywhere leaves a session nothing collects, which +the next pass finishes. With the directories removed first, a crash before the marker would let a +collector land the session again from its start, as a second chain under an id a receiver may still +hold. + +Every pass first finishes a retirement left unfinished, before anything lands, whatever the settings +say now, both off included. `asz parse` and `asz push` do the same before anything else. A request +that waits stays until it is sent, and the next pass after the send takes it away, and the owner +with it. + +Nothing of a retired session is written again. Each collector passes it over before it makes a +directory: Claude Code's transcripts, child agents and workflows, the plugin's records, Kiro's +sessions and their sub-executions, and provider bodies. A LangSmith run whose conversation is +retired is not collected and never waits for another. The metrics deriver and the parser do nothing +for it, and `asz push` sends none of its files or rounds; its waiting requests still go. `asz repack` +refuses a root that holds a retired session, since a new root without the marker would collect it +again, and an sd scenario build refuses a session retired in its root. The pass log says once a +process that a retired session's source was found and not collected, `asz sources` names it, and +`asz verify` counts the retired sessions. + +While a retirement is unfinished, the page lists and draws the conversation as long as its chain is +there, and once the chain is gone it does neither. A read of the page never makes anything of the +session again. + ### Removing by hand The session directory is the unit of retention: removing it purges everything collected for that @@ -566,11 +660,12 @@ with a `no rounds` error, and so does the request for the document behind the pa ### A scenario root -asz removes a collected session whole on its own in one case only: a session a `claude-code` -scenario build wrote. [Scenarios](../guides/scenario.md#removal-a-session-goes-once-it-is-sent) -describes the policy. A real session is never removed whole. Only the build writes the marker a -whole removal needs. [Retention](#retention), when it is set, removes the landed content of an idle -conversation, and keeps its structure, in any root but one with `_scenario/`: an sd build's +asz removes a collected session whole on its own in two cases, which are separate. One is +[retirement](#retirement), and a retired id is never collected again. The other is a session a +`claude-code` scenario build wrote, described here. +[Scenarios](../guides/scenario.md#removal-a-session-goes-once-it-is-sent) describes the policy. +Only the build writes the marker this removal needs. +[Retention](#retention), when it is set, runs in any root but one with `_scenario/`, an sd build's included. The build writes one marker for each session, `DIR/_source/.asz-scenario/.json`, and a diff --git a/docs/en/guides/command-line-internals.md b/docs/en/guides/command-line-internals.md index b764452..93eb3d2 100644 --- a/docs/en/guides/command-line-internals.md +++ b/docs/en/guides/command-line-internals.md @@ -37,15 +37,22 @@ leaving them for the next one. With `storage.retention.metadata_only_after` set, a pass then keeps only the metadata of every conversation idle for longer, sent or not: it removes the landed files and leaves a skeleton in -each one's place. Before it lands anything, every pass finishes a removal an earlier one left in -progress, whatever the setting says now. A scenario root keeps its own removal, and retention never -runs there. See [Retention](../formats/storage-root.md#retention). The startup lines say the +each one's place. With `storage.retention.retire_after` set, a pass first retires every +conversation idle for longer: everything of it is removed, and it is never collected again. Before +it lands anything, every pass finishes a removal or a retirement an earlier one left in progress, +whatever the settings say now. A scenario root keeps its own removal, and retention never runs +there. See [Retention](../formats/storage-root.md#retention). The startup lines say the retention: ```text retention: a conversation idle for 72h0m0s keeps only its metadata; its content is removed, and only a receiver that accepted it still has it +retention: a conversation idle for 2160h0m0s is retired: everything of it is removed, and it is never collected again, even when its runtime resumes it ``` +When both are the same, a conversation is retired with no metadata-only step first, and only the +second line is said. `asz parse` and `asz push` finish a retirement left in progress before they do +anything else. + When metrics are turned on after a conversation kept only its metadata, the skeletons give its tokens, and not whether an execution was a call to an MCP server. The first pass of a process says, for each session, how many distinct execution observations its skeletons keep that no pass counted. @@ -78,6 +85,7 @@ One line per pass, and a quiet pass prints nothing: | `rejected` | records and points a receiver took and rejected, absent when 0. It also gives an `error:` line, because they are never sent again. | | `removed` | scenario sessions removed, absent when 0. See [A scenario root](#a-scenario-root). | | `retention` | landed files `storage.retention.metadata_only_after` removed, absent when 0. See [Retention](../formats/storage-root.md#retention). | +| `retired` | conversations `storage.retention.retire_after` retired, absent when 0. See [Retirement](../formats/storage-root.md#retirement). | | `errors` | listed on standard error after the line | `busy` is not an error. Running two pipelines on one root is supported, and the process that holds @@ -360,8 +368,10 @@ build prints the policy after `config`, and a feed prints it in its header: remove : 24h after its last record, once sent ``` -`--remove` with `--format sd` is refused, because an sd build writes no source and its sessions are -never removed whole. Retention, when it is set, still applies to its root. `check` refuses `--remove`, and `build` refuses zero, a negative duration and +`--remove` with `--format sd` is refused, because an sd build writes no source and a scenario's +removal never removes its sessions. Retention, when it is set, still applies to its root, as to a +real one, and retirement removes a session there whole. A build of a session retired in its root is +refused. `check` refuses `--remove`, and `build` refuses zero, a negative duration and anything it cannot read. Each refusal exits with status 2 and writes nothing. ## push @@ -370,7 +380,9 @@ Sends every landed file and every round not yet sent to the OpenTelemetry logs r `export.otlp.endpoint`, one log record per file, then exits. One pass is all it does: a root that keeps growing is sent by `asz collect`, which pushes at the end of every period. This command is for a root that is already there, such as one copied from another machine or brought under a new -budget by `asz repack`. It never removes anything, even over a [scenario root](#a-scenario-root). +budget by `asz repack`. It starts no removal and no retirement, even over a [scenario +root](#a-scenario-root). It finishes a retirement a pass left unfinished, as every command that +writes to a root does first. One line for the pass: ```text diff --git a/docs/en/guides/scenario.md b/docs/en/guides/scenario.md index 2774c2d..59c15ba 100644 --- a/docs/en/guides/scenario.md +++ b/docs/en/guides/scenario.md @@ -152,12 +152,13 @@ written, so a feed's session stays about that long after it was written. A one-s old `--at` is already past its duration, and goes as soon as it is sent. `d` counts whole days, as `metrics.lookback` does. One run applies one policy to every session it writes. -The policy belongs to the scenario, never to the product. The configuration has no setting that -removes a session whole, and a session no build marked, which is every real Claude Code session, is -never removed whole. An sd build writes no source and no marker, so its sessions are never removed -whole, and `--remove` with `--format sd` is refused. `storage.retention`, a separate setting, keeps -only the metadata of an idle conversation in an unmarked root, an sd build's included. It never -runs over a root with `_scenario/`, which keeps its own removal. +The policy belongs to the scenario, never to the product. This removal never takes a session no +build marked, which is every real Claude Code session. An sd build writes no source and no marker, +so this removal never takes its sessions, and `--remove` with `--format sd` is refused. +`storage.retention`, a separate setting, keeps only the metadata of an idle conversation, and with +`retire_after` removes it whole and retires it, in an unmarked root, an sd build's included. A +build of a session retired in its root is refused. Retention never runs over a root with +`_scenario/`, which keeps its own removal. Nothing is removed that the pipeline cannot prove sent. Each of these keeps every marked session, and the pass says why once, on standard error, as `kept: every marked session: `: diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md index 5ed4864..5ec6673 100644 --- a/docs/en/setup/configuration.md +++ b/docs/en/setup/configuration.md @@ -17,7 +17,8 @@ curl -fsSL -o asz.yaml "https://raw.githubusercontent.com/apache/skywalking-ai-s | Key | Default | Meaning | | --- | --- | --- | | `root` | `./data` | Where collected data lands and conversations are assembled. A relative path is resolved from the working directory. | -| `retention.metadata_only_after` | `-1` | How long a conversation may be idle, with nothing new landed for it, before the root keeps only its metadata, to save disk: what was said and run is removed, and who did what, when, in what order and at what cost stays. New activity continues the same conversation. The removed content cannot be restored from the root; only a receiver that accepted it still has it. A duration such as `72h`, or days such as `30d`. It applies whether the conversation was sent or not, and whether an endpoint is set or not. `-1` is off, and only off keeps everything. See [Retention](../formats/storage-root.md#retention). | +| `retention.metadata_only_after` | `-1` | How long a conversation may be idle, with nothing new landed for it, before the root keeps only its metadata, to save disk: what was said and run is removed, and who did what, when, in what order and at what cost stays. New activity continues the same conversation. The removed content cannot be restored from the root; only a receiver that accepted it still has it. A duration such as `72h`, or days such as `30d`. It applies whether the conversation was sent or not, and whether an endpoint is set or not. `-1` is off. See [Retention](../formats/storage-root.md#retention). | +| `retention.retire_after` | `-1` | How long a conversation may be idle before everything of it is removed, its metadata too, and it is retired. A retired conversation is never collected again, even when its runtime resumes it, so set this longer than the runtime keeps its own history. It is measured on the same clock as `metadata_only_after`, and is at least as long. Equal to it, a conversation is retired with no metadata-only step first. Set with `metadata_only_after` off, or shorter than it, it is a configuration error. `-1` is off. Only both off keeps everything. See [Retirement](../formats/storage-root.md#retirement). | ## adapters diff --git a/internal/adapters/claudecodeprovider/seen.go b/internal/adapters/claudecodeprovider/seen.go index 29b88fe..4bc82f6 100644 --- a/internal/adapters/claudecodeprovider/seen.go +++ b/internal/adapters/claudecodeprovider/seen.go @@ -278,3 +278,35 @@ func Forget(z *storage.Zone, sessions []string, now time.Time) error { } return sn.save(z, now) } + +// Retire marks every line of the given sessions retired, keeping what each +// line says. Claude Code never deletes a body file, so these lines are what +// keeps each pass from reading a retired session's bodies again. A line goes +// once its file is gone, as any line of a session that is gone does. A +// retirement calls it once the sessions' directories are gone. It holds the +// provider lock already, in the order every command takes the locks, so +// Retire takes none, where Forget takes it itself. A table that does not +// exist has nothing to mark. +func Retire(z *storage.Zone, sessions []string, now time.Time) error { + if _, err := os.Stat(z.ProviderStatePath()); errors.Is(err, os.ErrNotExist) { + return nil + } + sn, err := loadSeen(z) + if err != nil { + return err + } + retired := map[string]bool{} + for _, s := range sessions { + retired[s] = true + } + for _, e := range sn.entries { + if retired[e.session] && e.state != stateRetired { + e.state = stateRetired + sn.dirty = true + } + } + if !sn.dirty { + return nil + } + return sn.save(z, now) +} diff --git a/internal/config/config.go b/internal/config/config.go index 5a3689a..a6004ee 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -209,6 +209,11 @@ type Retention struct { // its skeleton, which keeps who did what, when and in what order, and // none of what was said or run. See MetadataOnlyAfterDuration. MetadataOnlyAfter string `yaml:"metadata_only_after"` + // RetireAfter is how long a conversation may be idle before everything + // of it is removed, its metadata too, and it is retired: it is never + // collected again, even when its runtime resumes it. See + // RetireAfterDuration. + RetireAfter string `yaml:"retire_after"` } // MetadataOnlyAfterDuration is metadata_only_after as a duration: zero for -1 @@ -218,6 +223,35 @@ func (r Retention) MetadataOnlyAfterDuration() (time.Duration, error) { return retentionDuration("storage.retention.metadata_only_after", r.MetadataOnlyAfter) } +// RetireAfterDuration is retire_after as a duration, read as +// MetadataOnlyAfterDuration reads its own. +func (r Retention) RetireAfterDuration() (time.Duration, error) { + return retentionDuration("storage.retention.retire_after", r.RetireAfter) +} + +// Durations reads both levels and checks that they go together. Retirement +// removes the metadata level 1 keeps, so it comes no earlier than level 1, +// and it needs level 1 set: a root that keeps the metadata for ever would +// otherwise retire what it was told to keep. Equal levels retire a +// conversation with no level 1 before it. +func (r Retention) Durations() (metadataOnly, retire time.Duration, err error) { + if metadataOnly, err = r.MetadataOnlyAfterDuration(); err != nil { + return 0, 0, err + } + if retire, err = r.RetireAfterDuration(); err != nil { + return 0, 0, err + } + switch { + case retire > 0 && metadataOnly == 0: + return 0, 0, fmt.Errorf("config: storage.retention.retire_after is %s and metadata_only_after is off; set metadata_only_after too, at most %s", + strings.TrimSpace(r.RetireAfter), strings.TrimSpace(r.RetireAfter)) + case retire > 0 && retire < metadataOnly: + return 0, 0, fmt.Errorf("config: storage.retention.retire_after %s is shorter than metadata_only_after %s; it must be at least as long", + strings.TrimSpace(r.RetireAfter), strings.TrimSpace(r.MetadataOnlyAfter)) + } + return metadataOnly, retire, nil +} + func retentionDuration(name, value string) (time.Duration, error) { v := strings.TrimSpace(value) switch v { @@ -389,7 +423,7 @@ const ( // Default returns the configuration used when none is supplied. func Default() *Config { return &Config{ - Storage: Storage{Root: "./data", Retention: Retention{MetadataOnlyAfter: "-1"}}, + Storage: Storage{Root: "./data", Retention: Retention{MetadataOnlyAfter: "-1", RetireAfter: "-1"}}, Adapters: []Adapter{{ Name: AdapterClaudeCodeLocal, Enabled: true, @@ -508,6 +542,9 @@ func Load(path string) (*Config, error) { if loaded.Storage.Retention.MetadataOnlyAfter != "" { cfg.Storage.Retention.MetadataOnlyAfter = loaded.Storage.Retention.MetadataOnlyAfter } + if loaded.Storage.Retention.RetireAfter != "" { + cfg.Storage.Retention.RetireAfter = loaded.Storage.Retention.RetireAfter + } if len(loaded.Adapters) > 0 { cfg.Adapters = loaded.Adapters for i := range cfg.Adapters { @@ -710,7 +747,7 @@ func (c *Config) Validate() error { if _, err := c.Metrics.LookbackDuration(); err != nil { return err } - if _, err := c.Storage.Retention.MetadataOnlyAfterDuration(); err != nil { + if _, _, err := c.Storage.Retention.Durations(); err != nil { return err } if err := sessiondata.CheckWithholdable(c.View.Hide); err != nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 3eee641..0329622 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -514,3 +514,65 @@ func TestMetadataOnlyAfterDuration(t *testing.T) { t.Fatalf("a configuration with metadata_only_after soon validates: %v", err) } } + +// The two levels are read together. retire_after is off when it is left out, +// is at least metadata_only_after, and needs it set; equal levels are the +// same however each is written. A value that is not a duration is refused by +// the setting's whole name, as metadata_only_after's is. +func TestTheTwoRetentionLevelsGoTogether(t *testing.T) { + path := filepath.Join(t.TempDir(), "asz.yaml") + section := func(metadataOnly, retire string) string { + body := "storage:\n retention:\n" + if metadataOnly != "" { + body += " metadata_only_after: " + metadataOnly + "\n" + } + if retire != "" { + body += " retire_after: " + retire + "\n" + } + return body + } + day := 24 * time.Hour + for _, c := range []struct { + metadataOnly, retire string + want [2]time.Duration + }{ + {"-1", "-1", [2]time.Duration{0, 0}}, + {"", "", [2]time.Duration{0, 0}}, + {"30d", "", [2]time.Duration{30 * day, 0}}, + {"30d", "-1", [2]time.Duration{30 * day, 0}}, + {"30d", "90d", [2]time.Duration{30 * day, 90 * day}}, + {"30d", "720h", [2]time.Duration{30 * day, 30 * day}}, + {"720h", "30d", [2]time.Duration{30 * day, 30 * day}}, + } { + body := section(c.metadataOnly, c.retire) + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if err != nil { + t.Fatalf("%q: %v", body, err) + } + metadataOnly, retire, err := cfg.Storage.Retention.Durations() + if err != nil || metadataOnly != c.want[0] || retire != c.want[1] { + t.Errorf("%q read as %v and %v, %v; want %v", body, metadataOnly, retire, err, c.want) + } + } + for _, c := range []struct{ metadataOnly, retire, says string }{ + {"-1", "30d", "metadata_only_after is off"}, + {"", "30d", "metadata_only_after is off"}, + {"30d", "7d", "is shorter than metadata_only_after"}, + {"30d", "719h", "is shorter than metadata_only_after"}, + {"30d", "soon", "storage.retention.retire_after"}, + {"30d", "0", "storage.retention.retire_after"}, + {"30d", "-2", "storage.retention.retire_after"}, + {"30d", "99999999999d", "storage.retention.retire_after"}, + } { + body := section(c.metadataOnly, c.retire) + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + if _, err := Load(path); err == nil || !strings.Contains(err.Error(), c.says) { + t.Errorf("%q was taken, or refused without saying %q: %v", body, c.says, err) + } + } +} diff --git a/internal/export/otlp/sent.go b/internal/export/otlp/sent.go index d2f5990..e6eb9bf 100644 --- a/internal/export/otlp/sent.go +++ b/internal/export/otlp/sent.go @@ -25,6 +25,8 @@ import ( "sort" "strings" "time" + + "github.com/apache/skywalking-ai-sessionizer/internal/storage" ) // EndpointOf names a receiver the way push.state records it. @@ -104,10 +106,13 @@ func (s *Sent) SentTo(endpoint, rel, digest string) bool { // ForgetGone drops the pushed and rejected lines of every path that owned // reports true for and whose file is gone. It drops the owner line of a // session that owned reports true for, asked by its id, once nothing of the -// session is left: its directory and its conversation's are gone, and no -// line names a file of it. A removal that stopped part way keeps the owner, -// so what it left is still sent to the same receiver. Every endpoint line -// stays. The caller holds the root's export lock (storage.LockExport), +// session is left: its directory and its conversation's are gone, no line +// names a file of it, and no metrics request of it waits in the spool. A +// removal that stopped part way keeps the owner, so what it left is still +// sent to the same receiver. A retired session's requests stay until they +// are sent, and a request goes to its session's owner only: with the owner +// gone, it would go to whichever owner the configuration names then. Every +// endpoint line stays. The caller holds the root's export lock (storage.LockExport), // because a pusher rewrites the same file. // // A line is dropped only once its file is gone, and a pusher sends only files @@ -137,8 +142,13 @@ func ForgetGone(root string, owned func(rel string) bool, now time.Time) error { } } left := s.sessions() + waiting, err := spooled(root) + if err != nil { + return err + } for session := range s.owners { - if owned(session) && !left[session] && fileGone(root, session) && fileGone(root, conversationsDir+"/"+session) { + if owned(session) && !left[session] && !waiting[session] && + fileGone(root, session) && fileGone(root, conversationsDir+"/"+session) { delete(s.owners, session) changed = true } @@ -149,6 +159,19 @@ func ForgetGone(root string, owned func(rel string) bool, now time.Time) error { return s.save(path, now) } +// spooled names the sessions a metrics request in the spool belongs to. +func spooled(root string) (map[string]bool, error) { + files, err := storage.NewSpool(storage.NewZone(root)).List() + if err != nil { + return nil, err + } + out := map[string]bool{} + for _, f := range files { + out[f.Session] = true + } + return out, nil +} + // fileGone reports whether the file at rel under root surely does not exist. // Any other answer, a permission error included, keeps its line. func fileGone(root, rel string) bool { diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 7e6d770..b85daf6 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -265,7 +265,7 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) { } ss := state.session(session) receiptPath := func(lf storage.LandedFile) string { - return filepath.Join(d.Zone.SessionDir(session), "metrics", fmt.Sprintf("%06d.json", lf.Seq)) + return filepath.Join(ReceiptsDir(d.Zone, session), fmt.Sprintf("%06d.json", lf.Seq)) } // A pass whose progress was not saved may have published receipts for // files after one that waited for its grace. Their requests are in the diff --git a/internal/metrics/receipt.go b/internal/metrics/receipt.go index 54b22f5..5c78858 100644 --- a/internal/metrics/receipt.go +++ b/internal/metrics/receipt.go @@ -22,7 +22,11 @@ import ( "errors" "fmt" "io" + "io/fs" "os" + "path/filepath" + "strconv" + "strings" "time" "google.golang.org/protobuf/proto" @@ -131,3 +135,71 @@ func loadReceipt(path string, lf storage.LandedFile) (*receipt, error) { } return &r, nil } + +// ReceiptsDir is where a session's receipts are kept, one for each landed +// file derived, named by its sequence. +func ReceiptsDir(z *storage.Zone, session string) string { + return filepath.Join(z.SessionDir(session), "metrics") +} + +// HandOff writes again, from its receipt, every metrics request of a session +// the receiver has not answered, and makes each one durable. It reports how +// many it wrote. +// +// A retirement calls it before the session's directory goes, since the +// receipts are kept there and are what can make a request again. A request +// already on disk is written again too: its directory sync may have failed, +// and a later sync that succeeds does not prove it reached the disk. It is +// read-only, and on Windows a rename cannot replace a read-only file, so it +// is deleted first. A crash between the deletion and the write leaves the +// receipt, from which the retirement writes it again. +func HandOff(z *storage.Zone, session string, answered func(rel string) bool, sync func(dir string) error) (int, error) { + dir := ReceiptsDir(z, session) + items, err := os.ReadDir(dir) + if errors.Is(err, fs.ErrNotExist) { + return 0, nil + } + if err != nil { + return 0, err + } + spool := storage.NewSpool(z).Dir() + written := 0 + for _, it := range items { + digits, ok := strings.CutSuffix(it.Name(), ".json") + seq, err := strconv.ParseUint(digits, 10, 64) + if it.IsDir() || !ok || err != nil { + continue + } + data, err := os.ReadFile(filepath.Join(dir, it.Name())) + if err != nil { + return written, err + } + // An empty receipt is a reservation, which recorded no decision. + if len(data) == 0 { + continue + } + var r receipt + if err := json.Unmarshal(data, &r); err != nil { + return written, fmt.Errorf("metrics: read receipt %s: %w", it.Name(), err) + } + if r.Schema != 1 { + return written, fmt.Errorf("metrics: receipt %s has unsupported schema %d", it.Name(), r.Schema) + } + name := SpoolName(session, seq) + if len(r.Request) == 0 || answered(storage.SpoolDir+"/"+name) { + continue + } + path := filepath.Join(spool, name) + if err := storage.DeleteTree(path, nil); err != nil { + return written, err + } + if err := storage.WriteAtomicDurable(path, storage.PermLanded, func(w io.Writer) error { + _, err := w.Write(r.Request) + return err + }, sync); err != nil { + return written, err + } + written++ + } + return written, nil +} diff --git a/internal/retention/retention.go b/internal/retention/retention.go index d7b30cf..82ed213 100644 --- a/internal/retention/retention.go +++ b/internal/retention/retention.go @@ -15,10 +15,17 @@ // specific language governing permissions and limitations // under the License. -// Package retention keeps only the metadata of conversations that have been -// idle for longer than storage.retention.metadata_only_after: it removes their -// landed files, and leaves a skeleton in the place of each. A skeleton keeps every field the index is built from, so the -// conversation's index, and so its chain, go on as if nothing was removed. +// Package retention keeps less of a conversation the longer it is idle. +// +// Level 1, after storage.retention.metadata_only_after, keeps only its +// metadata: it removes the conversation's landed files and leaves a skeleton +// in the place of each. A skeleton keeps every field the index is built from, +// so the conversation's index, and so its chain, go on as if nothing was +// removed. +// +// Level 2, after storage.retention.retire_after, retires it: everything of +// it is removed, and a marker under _retired keeps it from ever being +// collected again. // // It is collector side: it reads the round chain by its format alone, as the // scenario remover does, because the chain's package is server side. @@ -57,6 +64,10 @@ type Remover struct { // nothing, though Recover still finishes a removal an earlier pass left in // progress. MetadataOnlyAfter time.Duration + // RetireAfter is how long a conversation must be idle before it is + // retired. Zero retires nothing, though Recover still finishes a + // retirement an earlier pass began. + RetireAfter time.Duration // Metrics says the pipeline derives metrics, so every file's derivation // must be committed before the file is removed: a removed file keeps no // execution record to derive from. @@ -72,6 +83,10 @@ type Remover struct { // Stop is called at each point a removal reaches, and a removal stops // there when it returns an error, as a crash would stop it. Tests set it. Stop func(Point) error + + // settled names the retired sessions this process found nothing left + // of, so a pass does not take the locks or read the state for them. + settled map[string]bool } // Point is a place in a removal: its phase, and how many files of the @@ -115,6 +130,9 @@ type Result struct { // Files counts the files. Removed []string Files int + // Retired names the sessions whose directories a retirement removed + // this pass. + Retired []string Errors []error } @@ -125,12 +143,12 @@ func (r *Remover) now() time.Time { return time.Now() } -// Recover finishes every removal an earlier pass left in progress. A pass -// runs it before anything lands, so no collector writes into a session whose -// files are being removed. It runs whatever the retention: a removal that -// was started is finished. +// Recover finishes every retirement and every removal an earlier pass left +// in progress. A pass runs it before anything lands, so no collector writes +// into a session whose files are being removed. It runs whatever the +// retention: what was started is finished. func (r *Remover) Recover() Result { - var res Result + res := r.RecoverRetired() sessions, err := sessionDirs(r.Zone.Root()) if err != nil { res.Errors = append(res.Errors, err) @@ -166,12 +184,13 @@ func (r *Remover) Recover() Result { return res } -// Pass removes the landed files of every session that may lose them now. -// It runs at the end of a pipeline pass, after the send. It also takes away -// what metrics derived from a removed file once the receiver has it. +// Pass retires every session that may be retired now, and removes the +// landed files of every other session that may lose them. It runs at the end +// of a pipeline pass, after the send. It also takes away what metrics derived +// from a removed file once the receiver has it. func (r *Remover) Pass(in Input) Result { var res Result - if r.MetadataOnlyAfter <= 0 { + if r.MetadataOnlyAfter <= 0 && r.RetireAfter <= 0 { return res } sessions, err := sessionDirs(r.Zone.Root()) @@ -180,13 +199,49 @@ func (r *Remover) Pass(in Input) Result { return res } r.exported(&res, func(sent *otlp.Sent) { + // Retirement is decided first. A session idle for retire_after is + // never given level 1, even while it waits to be retired, so with + // both levels equal no skeleton is written. A retired session is + // carried on by Recover, and nothing else writes to it. + skip := map[string]bool{} + var retiring []string + for _, id := range sessions { + if in.Busy[id] { + continue + } + retired, err := storage.Retired(r.Zone, id) + if err != nil || retired { + if err != nil { + res.Errors = append(res.Errors, fmt.Errorf("%s: %w", id, err)) + } + skip[id] = true + continue + } + if r.RetireAfter <= 0 { + continue + } + idle, err := r.idle(id) + if err != nil { + res.Errors = append(res.Errors, fmt.Errorf("%s: %w", id, err)) + skip[id] = true + continue + } + if idle { + skip[id] = true + retiring = append(retiring, id) + } + } + r.retire(&res, retiring, sent, true) + if r.MetadataOnlyAfter <= 0 { + return + } requests, err := storage.NewSpool(r.Zone).List() if err != nil { res.Errors = append(res.Errors, err) return } for _, id := range sessions { - if in.Busy[id] { + if in.Busy[id] || skip[id] { continue } n, err := r.locked(id, func() (int, error) { diff --git a/internal/retention/retire.go b/internal/retention/retire.go new file mode 100644 index 0000000..f2143bf --- /dev/null +++ b/internal/retention/retire.go @@ -0,0 +1,474 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package retention + +import ( + "errors" + "fmt" + "io" + "io/fs" + "os" + "path/filepath" + "strings" + + "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider" + "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp" + "github.com/apache/skywalking-ai-sessionizer/internal/metrics" + "github.com/apache/skywalking-ai-sessionizer/internal/storage" +) + +// The points a retirement reaches, in order, for Stop. A directory's point is +// reached with each place RemoveDir reports, "renamed" and "deleting", and +// with "removed" once the directory is gone. The state's point counts the +// state files rewritten. +const ( + PhaseRetireMarked = "retire marked" + PhaseRetireHanded = "retire handed off" + PhaseRetireSession = "retire session" + PhaseRetireChain = "retire chain" + PhaseRetireTidied = "retire tidied" + PhaseRetireState = "retire state" +) + +// errNotNow is an idle session that may not be retired yet. It is tried +// again on the next pass. +var errNotNow = errors.New("retention: the session may not be retired now") + +// idle says whether a session has been idle for retire_after. It is read +// from the header of the newest landed file or skeleton, the clock level 1 +// reads too, and never from a record's time, a file's time or a removal's. +func (r *Remover) idle(id string) (bool, error) { + files, err := storage.LandedFiles(r.Zone, id) + if err != nil || len(files) == 0 { + return false, err + } + at, err := landedAt(files[len(files)-1]) + if err != nil { + return false, err + } + return r.now().Sub(at) >= r.RetireAfter, nil +} + +// retirable says whether an idle session may be retired now: no removal of +// its files is in progress, its chain covers every landed file, and its +// metrics are derived when the pipeline derives them, so each file's request +// exists for the hand-off. Sending is not waited for. It is asked under the +// session's locks. +func (r *Remover) retirable(id string) (bool, error) { + files, err := storage.LandedFiles(r.Zone, id) + if err != nil || len(files) == 0 { + return false, err + } + if intent, err := storage.LoadRemoving(r.Zone.RemovingPath(id)); err != nil || intent != nil { + return false, err + } + through, parsed, err := chainThrough(r.Zone.Root(), id) + if err != nil || !parsed || through < files[len(files)-1].Seq { + return false, err + } + if r.Metrics { + return metrics.DerivedAll(r.Zone, id, files) + } + return true, nil +} + +// RecoverRetired carries on every retirement that an earlier pass or command +// began and did not finish, whatever the settings are now, both levels off +// included: a session whose marker is there is retired, and what is left of +// it goes. Every command that writes to a root runs it first, under the root +// lock. A retirement this process found finished is not looked at again: +// nothing writes to a retired session, so nothing of it comes back. +func (r *Remover) RecoverRetired() Result { + var res Result + ids, err := storage.RetiredSessions(r.Zone) + if err != nil { + res.Errors = append(res.Errors, err) + return res + } + var work []string + for _, id := range ids { + if !r.settled[id] { + work = append(work, id) + } + } + if len(work) == 0 { + return res + } + r.exported(&res, func(sent *otlp.Sent) { r.retire(&res, work, sent, false) }) + return res +} + +// retiredDirs are the directories of a session a retirement removes, and +// where RemoveDir moves each of them. +func (r *Remover) retiredDirs(id string) []string { + root := r.Zone.Root() + removed := filepath.Join(root, storage.RemovedDir) + return []string{ + r.Zone.SessionDir(id), filepath.Join(root, conversationsDir, id), + filepath.Join(removed, id), filepath.Join(removed, id+".chain"), + } +} + +// retire carries the retirement of each session as far as it goes, under the +// metrics and provider locks, in the order every command takes the locks: +// the marker, the spool hand-off, the directories and the spool tidy for each, +// then the root's state once for every session that got that far. decide says +// the sessions are idle and not retired yet, so whether each may be retired is +// asked under its locks. A session already retired is carried on, whatever it +// is now. +func (r *Remover) retire(res *Result, ids []string, sent *otlp.Sent, decide bool) { + if len(ids) == 0 { + return + } + root := r.Zone.Root() + spool := storage.NewSpool(r.Zone) + requests, err := spool.List() + if err != nil { + res.Errors = append(res.Errors, err) + return + } + // Each lock is taken only when there is something under it, since taking + // a lock makes its directory. The metrics lock is needed by a session + // whose receipts make requests, which may make the spool. + spoolThere, err := present(spool.Dir()) + if err != nil { + res.Errors = append(res.Errors, err) + return + } + needMetrics := spoolThere + for _, id := range ids { + if there, err := present(metrics.ReceiptsDir(r.Zone, id)); err == nil && there { + needMetrics = true + } + } + if needMetrics { + lock, err := storage.LockMetricsWait(root, 0) + if errors.Is(err, storage.ErrMetricsBusy) { + return + } + if err != nil { + res.Errors = append(res.Errors, err) + return + } + defer func() { _ = lock.Unlock() }() + } + providerThere, err := present(filepath.Join(root, "_provider")) + if err != nil { + res.Errors = append(res.Errors, err) + return + } + if providerThere { + lock, err := storage.LockProvider(root) + if errors.Is(err, storage.ErrProviderBusy) { + return + } + if err != nil { + res.Errors = append(res.Errors, err) + return + } + defer func() { _ = lock.Unlock() }() + } + + var done []string + for _, id := range ids { + removed, err := r.retireOne(id, sent, requests, decide) + // Its directories are gone even when what came after failed, and + // no later attempt finds them to say so. + if removed { + res.Retired = append(res.Retired, id) + } + if errors.Is(err, errBusy) || errors.Is(err, errNotNow) { + continue + } + if err != nil { + res.Errors = append(res.Errors, fmt.Errorf("%s: retiring: %w", id, err)) + continue + } + done = append(done, id) + } + if len(done) == 0 { + return + } + // Every directory a retirement changes is synced before the state forgets + // anything, the errors checked. A power loss that brought back a session's + // directory, its receipts with it, after push.state forgot what the + // receiver has, would make those requests again and send them, perhaps + // under another owner. One that brought back a deleted request would leave + // it with no line and no owner. Every attempt syncs them, not only one + // that changed something: an earlier attempt may have, and stopped on a + // failed sync. + for _, dir := range []string{root, filepath.Join(root, conversationsDir), filepath.Join(root, storage.RemovedDir), spool.Dir()} { + there, err := present(dir) + if err != nil { + res.Errors = append(res.Errors, err) + return + } + if !there { + continue + } + if err := r.syncDir(dir); err != nil { + res.Errors = append(res.Errors, fmt.Errorf("%s may not survive a power loss, so the retirement stops here: %w", dir, err)) + return + } + } + // A request is waiting when the receiver does not have it. The spool is + // read again for it, since the hand-off may have made a request again. + left, err := spool.List() + if err != nil { + res.Errors = append(res.Errors, err) + return + } + waiting := map[string]bool{} + for _, f := range left { + if _, answered := sent.Digest(storage.SpoolDir + "/" + filepath.Base(f.Path)); !answered { + waiting[f.Session] = true + } + } + + // Step 4, once for all of them. Each rewrites one state file, and a line + // goes only once what it names is gone, so no file is derived twice. The + // owner stays while a request of the session waits in the spool. + now := r.now() + if err := otlp.ForgetGone(root, owned(done), now); err != nil { + res.Errors = append(res.Errors, err) + return + } + if err := r.stop(PhaseRetireState, 1); err != nil { + res.Errors = append(res.Errors, err) + return + } + if needMetrics { + if err := metrics.Forget(r.Zone, done, now); err != nil { + res.Errors = append(res.Errors, err) + return + } + } + if err := r.stop(PhaseRetireState, 2); err != nil { + res.Errors = append(res.Errors, err) + return + } + if providerThere { + if err := claudecodeprovider.Retire(r.Zone, done, now); err != nil { + res.Errors = append(res.Errors, err) + return + } + } + if err := r.stop(PhaseRetireState, 3); err != nil { + res.Errors = append(res.Errors, err) + return + } + if r.settled == nil { + r.settled = map[string]bool{} + } + for _, id := range done { + if !waiting[id] { + r.settled[id] = true + } + } +} + +// retireOne is steps 1 to 3 and the spool tidy for one session, and reports +// whether any directory of it was there to remove, also when something after +// the directories failed. +// +// The marker goes first and is made durable before anything is deleted: +// after it, a crash anywhere leaves a session that nothing collects, and the +// next pass finishes it. It is written again by every attempt that deletes +// anything, since an earlier attempt may have renamed it and stopped before +// its sync, and a later sync that succeeds does not prove the first one's +// data reached the disk. +func (r *Remover) retireOne(id string, sent *otlp.Sent, requests []storage.SpoolFile, decide bool) (bool, error) { + root := r.Zone.Root() + dirs := r.retiredDirs(id) + sessionLock, err := lockIfThere(dirs[0], storage.LockSessionToRemove, storage.ErrSessionBusy) + if err != nil { + return false, err + } + defer func() { unlock(&sessionLock) }() + chainLock, err := lockIfThere(dirs[1], storage.LockChain, storage.ErrChainBusy) + if err != nil { + return false, err + } + defer func() { unlock(&chainLock) }() + if decide { + if now, err := r.retirable(id); err != nil || !now { + if err == nil { + err = errNotNow + } + return false, err + } + } + + left := false + for _, dir := range dirs { + there, err := present(dir) + if err != nil { + return false, err + } + left = left || there + } + var answered []string + for _, f := range requests { + if _, ok := sent.Digest(storage.SpoolDir + "/" + filepath.Base(f.Path)); ok && f.Session == id { + answered = append(answered, f.Path) + } + } + if !left && len(answered) == 0 && !decide { + return false, nil + } + + // Step 1. + if err := r.mark(id); err != nil { + return false, err + } + if err := r.stop(PhaseRetireMarked, 0); err != nil { + return false, err + } + + // Step 2, while the session's directory holds its receipts. A directory + // renamed away was renamed after its hand-off was durable. + answeredBy := func(rel string) bool { + _, ok := sent.Digest(rel) + return ok + } + written, err := metrics.HandOff(r.Zone, id, answeredBy, r.syncDir) + if err != nil { + return false, err + } + // The spool may be new, and its own name is in the root. + if written > 0 { + if err := r.syncDir(root); err != nil { + return false, fmt.Errorf("the spool may not survive a power loss, so the retirement stops here: %w", err) + } + } + if err := r.stop(PhaseRetireHanded, written); err != nil { + return false, err + } + + // Step 3. Each directory goes in one rename, its lock released first: + // Windows cannot rename a directory that holds an open lock. The two + // directories each rename changed are synced, the errors checked, before + // anything moved is deleted. + for i, held := range []**storage.SessionLock{&sessionLock, &chainLock} { + dir, target := dirs[i], dirs[i+2] + phase := []string{PhaseRetireSession, PhaseRetireChain}[i] + unlock(held) + if err := storage.RemoveDirDurable(root, dir, filepath.Base(target), func(where string) error { + return r.stop(phase+" "+where, 0) + }, r.syncDir); err != nil { + return false, err + } + if err := r.stop(phase+" removed", 0); err != nil { + return false, err + } + } + + // The spool tidy: every request of the session the receiver has, a + // partly rejected one included, since a pusher does not send it again. + // The spool is synced with the directories, before the state. + if len(answered) > 0 { + for _, path := range answered { + if err := storage.DeleteTree(path, nil); err != nil { + return left, err + } + } + if err := r.stop(PhaseRetireTidied, len(answered)); err != nil { + return left, err + } + } + return left, nil +} + +// mark writes a session's retired marker and makes it durable: the marker's +// directory and the root are synced, every error checked. A marker already +// there is written again over itself, never deleted first, so it is never +// absent. It is writable, since on Windows a rename cannot replace a +// read-only file. +func (r *Remover) mark(id string) error { + path := r.Zone.RetiredPath(id) + if err := storage.WriteAtomicDurable(path, storage.PermState, func(io.Writer) error { return nil }, r.syncDir); err != nil { + return err + } + if err := r.syncDir(r.Zone.Root()); err != nil { + return fmt.Errorf("%s may not survive a power loss, so the retirement stops here: %w", storage.RetiredDir, err) + } + return nil +} + +// owned says which lines of push.state belong to the sessions: their landed +// files, their conversation's rounds, their metrics requests, and their owner +// lines, which are named by the session's id alone. +func owned(ids []string) func(rel string) bool { + set := map[string]bool{} + for _, id := range ids { + set[id] = true + } + return func(rel string) bool { + first, rest, ok := strings.Cut(rel, "/") + switch { + case !ok: + return set[rel] + case first == conversationsDir: + conv, _, _ := strings.Cut(rest, "/") + return set[conv] + case first == storage.SpoolDir: + session, ok := metrics.SpoolOwner(rest) + return ok && set[session] + } + return set[first] + } +} + +// lockIfThere takes a directory's lock only when the directory is there, +// since taking a lock makes its directory. +func lockIfThere(dir string, lock func(string) (*storage.SessionLock, error), busy error) (*storage.SessionLock, error) { + fi, err := os.Lstat(dir) + switch { + case errors.Is(err, fs.ErrNotExist): + return nil, nil + case err != nil: + return nil, err + case !fi.IsDir(): + return nil, fmt.Errorf("%s is a symbolic link or not a directory", dir) + } + l, err := lock(dir) + if errors.Is(err, busy) { + return nil, errBusy + } + return l, err +} + +func unlock(held **storage.SessionLock) { + if *held != nil { + _ = (*held).Unlock() + *held = nil + } +} + +// present reports whether a path is there. Anything other than its not being +// there is an error. +func present(path string) (bool, error) { + _, err := os.Lstat(path) + switch { + case err == nil: + return true, nil + case errors.Is(err, fs.ErrNotExist): + return false, nil + } + return false, err +} diff --git a/internal/storage/removed.go b/internal/storage/removed.go index f9a37dd..54dd270 100644 --- a/internal/storage/removed.go +++ b/internal/storage/removed.go @@ -58,6 +58,21 @@ const RemovedDir = "_removed" // with "deleting" after the first entry of the moved tree is deleted. An // error from it stops RemoveDir there, as a crash would. The tests use it. func RemoveDir(root, dir, name string, hook func(where string) error) error { + return removeDir(root, dir, name, hook, nil) +} + +// RemoveDirDurable is RemoveDir for a removal something is about to depend on +// surviving a power loss. After the rename, both directories it changed are +// synced with sync, and a failure is returned before anything moved is +// deleted. The first sync after a failed writeback is the one that knows. +func RemoveDirDurable(root, dir, name string, hook func(where string) error, sync func(dir string) error) error { + if sync == nil { + sync = SyncDir + } + return removeDir(root, dir, name, hook, sync) +} + +func removeDir(root, dir, name string, hook func(where string) error, sync func(dir string) error) error { if name == "" || name == "." || name == ".." || strings.ContainsAny(name, `/\`) { return fmt.Errorf("storage: %q is not a name for an entry of %s", name, RemovedDir) } @@ -103,11 +118,19 @@ func RemoveDir(root, dir, name string, hook func(where string) error) error { if err := os.Rename(dir, target); err != nil { return err } - // Either side of the rename is a whole directory, so a crash that loses - // the rename is safe. Syncing both directories only narrows that window, - // and a failure to sync leaves it as it was. - _ = syncDir(filepath.Dir(dir)) - _ = syncDir(filepath.Dir(target)) + // Either side of the rename is a whole directory, so for RemoveDir a + // crash that loses the rename is safe. Syncing both directories only + // narrows that window, and a failure to sync leaves it as it was. + if sync == nil { + _ = syncDir(filepath.Dir(dir)) + _ = syncDir(filepath.Dir(target)) + } else { + for _, changed := range []string{filepath.Dir(dir), filepath.Dir(target)} { + if err := sync(changed); err != nil { + return fmt.Errorf("storage: the move of %s may not survive a power loss: %w", dir, err) + } + } + } if hook != nil { if err := hook("renamed"); err != nil { return err diff --git a/internal/view/cache_test.go b/internal/view/cache_test.go index bda3a82..7e692f2 100644 --- a/internal/view/cache_test.go +++ b/internal/view/cache_test.go @@ -23,6 +23,7 @@ import ( "testing" "github.com/apache/skywalking-ai-sessionizer/internal/storage" + "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow" ) // TestAFoldIsKeptByWhatItReachedNotByWhatWasListed. @@ -53,3 +54,56 @@ func TestAFoldIsKeptByWhatItReachedNotByWhatWasListed(t *testing.T) { t.Fatal("a fold that failed was put in the cache; the next read would never try again") } } + +// TestListForgetsAConversationRemovedByAnotherProcess. A separate collector +// can retire a conversation while no page is open. Listing the root must +// release its stored fold without another Load or an explicit Forget. +func TestListForgetsAConversationRemovedByAnotherProcess(t *testing.T) { + root := t.TempDir() + s := New(storage.NewZone(root), nil) + for _, id := range []string{"removed", "kept"} { + w, err := sessionflow.NewWriter(sessionflow.Header{ + Conversation: id, Session: id, Round: 1, Parser: "v1", Policy: "v1", + FromSeq: 1, InputDigest: "d0", + }) + if err != nil { + t.Fatal(err) + } + data, digest, err := w.Close() + if err != nil { + t.Fatal(err) + } + if _, err := sessionflow.OpenChain(root, id).Publish(1, digest, data); err != nil { + t.Fatal(err) + } + if _, err := s.Load(id); err != nil { + t.Fatal(err) + } + if !s.Kept(id) { + t.Fatalf("%s was not kept after loading", id) + } + } + if err := storage.RemoveDir(root, sessionflow.OpenChain(root, "removed").Dir(), "removed.chain", nil); err != nil { + t.Fatal(err) + } + ids, err := s.List() + if err != nil || len(ids) != 1 || ids[0] != "kept" { + t.Fatalf("listed %v, %v; want only the remaining conversation", ids, err) + } + if s.Kept("removed") { + t.Fatal("the removed conversation is still kept after listing") + } + if !s.Kept("kept") { + t.Fatal("the remaining conversation was forgotten") + } + // A missing parent directory is an empty list too. + if err := storage.DeleteTree(filepath.Join(root, "_conversations"), nil); err != nil { + t.Fatal(err) + } + if ids, err := s.List(); err != nil || len(ids) != 0 { + t.Fatalf("listed %v, %v after the parent directory went", ids, err) + } + if s.Kept("kept") { + t.Fatal("the last conversation is still kept after its parent directory went") + } +} diff --git a/internal/view/view.go b/internal/view/view.go index 9eba8f6..73906a2 100644 --- a/internal/view/view.go +++ b/internal/view/view.go @@ -226,6 +226,14 @@ func (s *Server) Forget(id string) { s.mu.Unlock() } +// Kept reports whether the page keeps a conversation's fold in memory. +func (s *Server) Kept(id string) bool { + s.mu.Lock() + defer s.mu.Unlock() + _, ok := s.loaded[id] + return ok +} + // Time returns when a node happened, or 0 when nothing observed it. // // A node with no reference has no observed time, and a zero must render as @@ -323,20 +331,18 @@ func (c *Conversation) Streams() []*sessionflow.Node { return out } -// List returns every conversation in the zone that has rounds. // Root is the storage root the page reads. func (s *Server) Root() string { return s.zone.Root() } +// List returns every conversation in the zone that has rounds. func (s *Server) List() ([]string, error) { base := filepath.Join(s.zone.Root(), "_conversations") items, err := os.ReadDir(base) - if err != nil { - if os.IsNotExist(err) { - return nil, nil - } + if err != nil && !os.IsNotExist(err) { return nil, err } var out []string + listed := make(map[string]bool, len(items)) for _, d := range items { if !d.IsDir() { continue @@ -345,8 +351,20 @@ func (s *Server) List() ([]string, error) { // file, so an entry in the directory is not yet a round. if rounds, rerr := sessionflow.OpenChain(s.zone.Root(), d.Name()).List(); rerr == nil && len(rounds) > 0 { out = append(out, d.Name()) + listed[d.Name()] = true + } + } + // Another process can retire a conversation. The root watcher lists + // even with no page open, so its next pass releases the stored fold. + // The lock is held for this alone, so a read of the page does not wait + // for the listing. + s.mu.Lock() + for id := range s.loaded { + if !listed[id] { + delete(s.loaded, id) } } + s.mu.Unlock() sort.Strings(out) return out, nil } diff --git a/tests/chain/retire_test.go b/tests/chain/retire_test.go new file mode 100644 index 0000000..6c7ed9f --- /dev/null +++ b/tests/chain/retire_test.go @@ -0,0 +1,1190 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package chain_test + +import ( + "bytes" + "errors" + "fmt" + "io/fs" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sort" + "strings" + "testing" + "time" + + "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp" + "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest" + "github.com/apache/skywalking-ai-sessionizer/internal/metrics" + "github.com/apache/skywalking-ai-sessionizer/internal/retention" + "github.com/apache/skywalking-ai-sessionizer/internal/scenario" + "github.com/apache/skywalking-ai-sessionizer/internal/storage" + "github.com/apache/skywalking-ai-sessionizer/internal/view" +) + +// retirementOf is a remover whose levels are both past every file a test +// lands, so it retires every session it may, with no skeleton first. +func retirementOf(r *sentRoot) *retention.Remover { + return &retention.Remover{Zone: r.zone, MetadataOnlyAfter: time.Hour, RetireAfter: time.Hour, Metrics: true, Now: later} +} + +// retiredWhole checks that nothing of a retired session is left but its +// marker and the metrics requests named, which wait to be sent: no +// directory of it, here or under _removed; no other request; no line of +// metrics.state naming it; no line of push.state naming it, its owner aside +// while a request waits; and every line of the provider's table that names +// it retired. +func retiredWhole(t *testing.T, z *storage.Zone, session string, waiting ...string) { + t.Helper() + root := z.Root() + if retired, err := storage.Retired(z, session); err != nil || !retired { + t.Fatalf("%s has no marker: %v", session, err) + } + for _, dir := range []string{z.SessionDir(session), filepath.Join(root, "_conversations", session), + filepath.Join(root, storage.RemovedDir, session), filepath.Join(root, storage.RemovedDir, session+".chain")} { + if present(dir) { + t.Fatalf("%s is left of the retired session", dir) + } + } + if left := spoolOf(t, root, session); strings.Join(left, ",") != strings.Join(waiting, ",") { + t.Fatalf("the spool holds %v of the retired session, want %v", left, waiting) + } + if data, err := os.ReadFile(filepath.Join(root, storage.SpoolDir, metrics.StateFile)); err == nil && bytes.Contains(data, []byte(session)) { + t.Fatalf("metrics.state still names the retired session:\n%s", data) + } else if err != nil && !errors.Is(err, fs.ErrNotExist) { + t.Fatal(err) + } + sent, err := otlp.LoadSent(root) + if err != nil { + t.Fatal(err) + } + if owner := sent.Owner(session); owner != "" && len(waiting) == 0 { + t.Fatalf("push.state keeps the owner %q with no request waiting", owner) + } + data, err := os.ReadFile(filepath.Join(root, otlp.StateFile)) + if err != nil && !errors.Is(err, fs.ErrNotExist) { + t.Fatal(err) + } + for _, line := range strings.Split(string(data), "\n") { + ownerKept := len(waiting) > 0 && strings.HasPrefix(line, "owner "+session+" ") + if strings.Contains(line, session) && !ownerKept { + t.Fatalf("push.state still names the retired session: %q", line) + } + } + data, err = os.ReadFile(z.ProviderStatePath()) + if err != nil && !errors.Is(err, fs.ErrNotExist) { + t.Fatal(err) + } + for _, line := range strings.Split(string(data), "\n") { + if f := strings.Split(line, "\t"); len(f) == 10 && f[5] == session && f[4] != "retired" { + t.Fatalf("the provider's table keeps a line of the retired session that is not retired: %q", line) + } + } +} + +// noRecordOf fails when the receiver holds a log record of the session. +func noRecordOf(t *testing.T, rcv *otlptest.Receiver, session string) { + t.Helper() + for _, rec := range otlptest.Records(rcv.Requests()) { + if attrs := otlptest.Attrs(rec.Attributes); attrs["asz.session"] == session { + t.Fatalf("%s of the retired session was sent", attrs["asz.file"]) + } + } +} + +// TestEveryScenarioRetiredWritesNothingAgain. Every scenario is collected, +// derived, parsed and sent through its checkpoints, and then retired whole: +// nothing of any of its sessions is left but the marker. Its sources are +// then offered again with their cursors gone, which a collector that did not +// ask would land again from the start, and a second pass of every writer +// writes nothing of it and sends nothing of it. +func TestEveryScenarioRetiredWritesNothingAgain(t *testing.T) { + paths, err := filepath.Glob(filepath.Join("..", "scenarios", "*.yaml")) + if err != nil { + t.Fatal(err) + } + ran := 0 + for _, path := range paths { + if strings.HasSuffix(path, ".expect.yaml") { + continue + } + name := strings.TrimSuffix(filepath.Base(path), ".yaml") + sc, err := scenario.Load(path) + if err != nil { + t.Fatal(err) + } + t.Run(name, func(t *testing.T) { + r := newSentRoot(t) + r.maxDelta = budgetOf(t, name) + k := 0 + for _, cp := range append(sc.Checkpoints(), "") { + r.build(sc, scenario.Options{Through: cp}) + r.pipelineAt(landedAt(k)) + k++ + } + sessions := sessionsOf(t, r.dir) + if len(sessions) == 0 { + t.Skip("the scenario lands no session") + } + res := retirementOf(r).Pass(retention.Input{}) + if len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + if strings.Join(res.Retired, ",") != strings.Join(sessions, ",") { + t.Fatalf("retired %v, want every session %v", res.Retired, sessions) + } + for _, s := range sessions { + retiredWhole(t, r.zone, s) + } + + r.rcv.Reset() + r.pipelineAt(landedAt(k)) + for _, rem := range []*retention.Remover{retirementOf(r), {Zone: r.zone}} { + if res := rem.Recover(); len(res.Errors) > 0 || len(res.Retired) > 0 { + t.Fatalf("a recovery of finished retirements did something: %+v", res) + } + } + for _, s := range sessions { + retiredWhole(t, r.zone, s) + noRecordOf(t, r.rcv, s) + } + if n := len(r.rcv.MetricsRequests()); n != 0 { + t.Fatalf("%d metrics requests were sent after the retirement", n) + } + }) + ran++ + } + if ran == 0 { + t.Fatal("no scenario ran") + } +} + +// sessionFiles lists the session's landed files and skeletons by name. +func sessionFiles(t *testing.T, z *storage.Zone, session string) (landed, skeletons int) { + t.Helper() + files, err := storage.LandedFiles(z, session) + if err != nil { + t.Fatal(err) + } + for _, lf := range files { + if lf.Skeleton { + skeletons++ + } else { + landed++ + } + } + return landed, skeletons +} + +// TestRetirementFollowsTheNewestLandingTime. Idle is read from the header of +// the session's newest landed file, by asz's clock, whatever the record +// times, the file times and a skeleton's removal time say, and whether its +// sources are there or not. +// - Idle for level 1 and not for level 2: only the metadata is kept. +// - Then idle for level 2: the session, all skeletons now, is retired. Its +// skeletons were written later than its newest landing, and that changes +// nothing. +// - Idle for both on its first pass, the levels apart or equal: it is +// retired with no skeleton written first. +func TestRetirementFollowsTheNewestLandingTime(t *testing.T) { + newest := landedAt(1) + idle := func(d time.Duration) func() time.Time { return func() time.Time { return newest.Add(d) } } + growthRoot := func(t *testing.T) (*sentRoot, string) { + r := newSentRoot(t) + b := growthAt(t, r, 2, func(k int) { r.pipelineAt(landedAt(k)) }) + // The record times are a month before the landing, and the file + // times are years before it. + files, err := storage.LandedFiles(r.zone, b.Session) + if err != nil { + t.Fatal(err) + } + old := time.Date(2001, 1, 1, 0, 0, 0, 0, time.UTC) + for _, lf := range files { + if err := os.Chtimes(lf.Path, old, old); err != nil { + t.Fatal(err) + } + } + return r, b.Session + } + + t.Run("level 1, then level 2", func(t *testing.T) { + r, session := growthRoot(t) + rem := &retention.Remover{Zone: r.zone, MetadataOnlyAfter: time.Hour, RetireAfter: 2 * time.Hour, Metrics: true} + rem.Now = idle(2*time.Hour - time.Minute) + if res := rem.Pass(retention.Input{}); len(res.Errors) > 0 || len(res.Retired) > 0 || res.Files == 0 { + t.Fatalf("idle for level 1 only: %+v", res) + } + if landed, skeletons := sessionFiles(t, r.zone, session); landed != 0 || skeletons == 0 { + t.Fatalf("%d landed files and %d skeletons after level 1", landed, skeletons) + } + rem.Now = idle(2 * time.Hour) + if res := rem.Pass(retention.Input{}); len(res.Errors) > 0 || strings.Join(res.Retired, ",") != session { + t.Fatalf("idle for level 2: %+v", res) + } + retiredWhole(t, r.zone, session) + }) + + for _, c := range []struct { + name string + metadataOnly, level time.Duration + }{ + {"apart", time.Hour, 2 * time.Hour}, + {"equal", 2 * time.Hour, 2 * time.Hour}, + } { + t.Run("both on the first pass, "+c.name, func(t *testing.T) { + r, session := growthRoot(t) + if err := os.RemoveAll(r.source); err != nil { + t.Fatal(err) + } + rem := &retention.Remover{Zone: r.zone, MetadataOnlyAfter: c.metadataOnly, RetireAfter: c.level, Metrics: true} + if c.metadataOnly == c.level { + // Equal levels do nothing a second short of either. + rem.Now = idle(c.level - time.Second) + if res := rem.Pass(retention.Input{}); len(res.Errors) > 0 || len(res.Retired) > 0 || res.Files != 0 { + t.Fatalf("a second short of both levels: %+v", res) + } + } + skeletonsAtRetirement := -1 + rem.Stop = func(p retention.Point) error { + if p.Phase == retention.PhaseRetireMarked { + _, skeletonsAtRetirement = sessionFiles(t, r.zone, session) + } + return nil + } + rem.Now = idle(c.level) + res := rem.Pass(retention.Input{}) + if len(res.Errors) > 0 || strings.Join(res.Retired, ",") != session || res.Files != 0 { + t.Fatalf("idle for both: %+v", res) + } + if skeletonsAtRetirement != 0 { + t.Fatalf("%d skeletons were written before the retirement", skeletonsAtRetirement) + } + retiredWhole(t, r.zone, session) + }) + } +} + +// TestRetirementWaitsForWhatItNeeds. A session idle past retire_after is not +// retired while a landed file is not parsed, its metrics are not derived, a +// removal of its files is in progress, or its parse failed this pass. Level 1 +// is not given to it meanwhile either. Once none holds, it is retired. +func TestRetirementWaitsForWhatItNeeds(t *testing.T) { + waits := []struct { + name string + hold func(r *sentRoot, session string) retention.Input + clear func(r *sentRoot, session string) + }{ + {"not parsed", func(r *sentRoot, _ string) retention.Input { + r.collectAt(landedAt(2)) + r.derive() + return retention.Input{} + }, func(r *sentRoot, _ string) { r.parse() }}, + {"metrics not derived", func(r *sentRoot, _ string) retention.Input { + r.collectAt(landedAt(2)) + r.parse() + return retention.Input{} + }, func(r *sentRoot, _ string) { r.derive() }}, + {"a removal in progress", func(r *sentRoot, session string) retention.Input { + files, err := storage.LandedFiles(r.zone, session) + if err != nil { + r.t.Fatal(err) + } + intent := &storage.Removing{Generation: 1, Through: files[len(files)-1].Seq, Phase: storage.RemovingSkeletons} + if err := storage.SaveRemoving(r.zone.RemovingPath(session), intent); err != nil { + r.t.Fatal(err) + } + return retention.Input{} + }, func(r *sentRoot, _ string) { + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + r.t.Fatal(res.Errors) + } + }}, + {"busy", func(_ *sentRoot, session string) retention.Input { + return retention.Input{Busy: map[string]bool{session: true}} + }, func(*sentRoot, string) {}}, + } + for _, w := range waits { + t.Run(w.name, func(t *testing.T) { + r := newSentRoot(t) + sc, err := scenario.Load(filepath.Join("..", "scenarios", "growth.yaml")) + if err != nil { + t.Fatal(err) + } + cps := append(sc.Checkpoints(), "") + r.build(sc, scenario.Options{Through: cps[0]}) + r.pipelineAt(landedAt(1)) + b := r.build(sc, scenario.Options{Through: cps[1]}) + in := w.hold(r, b.Session) + landedBefore, skeletonsBefore := sessionFiles(t, r.zone, b.Session) + rem := retirementOf(r) + rem.MetadataOnlyAfter = time.Minute + if res := rem.Pass(in); len(res.Errors) > 0 || len(res.Retired) > 0 { + t.Fatalf("it was retired, or failed: %+v", res) + } + if retired, _ := storage.Retired(r.zone, b.Session); retired { + t.Fatal("it has a marker") + } + if landed, skeletons := sessionFiles(t, r.zone, b.Session); landed != landedBefore || skeletons != skeletonsBefore { + t.Fatalf("level 1 ran while it waited to be retired: %d landed and %d skeletons, %d and %d before", + landed, skeletons, landedBefore, skeletonsBefore) + } + w.clear(r, b.Session) + if res := rem.Pass(retention.Input{}); len(res.Errors) > 0 || strings.Join(res.Retired, ",") != b.Session { + t.Fatalf("once nothing holds it: %+v", res) + } + }) + } +} + +// retiringRoot builds a scenario through its checkpoints with a pipeline pass +// after each build, and sends every pass but the last. Its session then holds +// metrics requests the receiver has, and requests it does not. It returns the +// root, the session, and the requests not sent, by name, with their bytes. +func retiringRoot(t *testing.T, name string) (*sentRoot, string, map[string][]byte) { + t.Helper() + r := newSentRoot(t) + sc, err := scenario.Load(filepath.Join("..", "scenarios", name+".yaml")) + if err != nil { + t.Fatal(err) + } + r.maxDelta = budgetOf(t, name) + cps := append(sc.Checkpoints(), "") + var b *scenario.Built + for k, cp := range cps { + b = r.build(sc, scenario.Options{Through: cp}) + r.collectAt(landedAt(k)) + r.derive() + r.parse() + if k < len(cps)-1 { + r.push() + } + } + sent, err := otlpSent(r) + if err != nil { + t.Fatal(err) + } + waiting, answered := map[string][]byte{}, 0 + for _, name := range spoolOf(t, r.dir, b.Session) { + if _, ok := sent.Digest(storage.SpoolDir + "/" + name); ok { + answered++ + continue + } + data, err := os.ReadFile(filepath.Join(r.dir, storage.SpoolDir, name)) + if err != nil { + t.Fatal(err) + } + waiting[name] = data + } + if answered == 0 || len(waiting) == 0 { + t.Fatalf("%d requests sent and %d not; the test needs both", answered, len(waiting)) + } + return r, b.Session, waiting +} + +// names lists a set of requests by name, in order. +func names(requests map[string][]byte) []string { + out := make([]string, 0, len(requests)) + for name := range requests { + out = append(out, name) + } + sort.Strings(out) + return out +} + +// sameRequests checks the spool holds each request with the bytes given. +func sameRequests(t *testing.T, dir string, requests map[string][]byte) { + t.Helper() + for name, want := range requests { + got, err := os.ReadFile(filepath.Join(dir, storage.SpoolDir, name)) + if err != nil || !bytes.Equal(got, want) { + t.Fatalf("the request %s is not kept as it was: %v", name, err) + } + } +} + +// TestRetirementFinishesAfterACrash. A retirement stopped at each of its +// points, as a crash would stop it, leaves a session nothing writes to: a +// pass of every collector, the deriver and the parser in between changes +// nothing of what is left. The next recovery finishes it, with both levels +// off, and a second recovery has nothing to do. The requests the receiver +// did not have stay, as they were, until they are sent; a recovery after the +// send takes them away, and the owner with them. +func TestRetirementFinishesAfterACrash(t *testing.T) { + base, session, waiting := retiringRoot(t, "provider-bodies-long") + const anyDone = -1 + stops := []retention.Point{ + {Phase: retention.PhaseRetireMarked, Done: anyDone}, + {Phase: retention.PhaseRetireHanded, Done: anyDone}, + {Phase: retention.PhaseRetireSession + " renamed", Done: anyDone}, + {Phase: retention.PhaseRetireSession + " deleting", Done: anyDone}, + {Phase: retention.PhaseRetireSession + " removed", Done: anyDone}, + {Phase: retention.PhaseRetireChain + " renamed", Done: anyDone}, + {Phase: retention.PhaseRetireChain + " deleting", Done: anyDone}, + {Phase: retention.PhaseRetireChain + " removed", Done: anyDone}, + {Phase: retention.PhaseRetireTidied, Done: anyDone}, + {Phase: retention.PhaseRetireState, Done: 1}, + {Phase: retention.PhaseRetireState, Done: 2}, + {Phase: retention.PhaseRetireState, Done: 3}, + } + for _, at := range stops { + t.Run(fmt.Sprintf("%s %d", at.Phase, at.Done), func(t *testing.T) { + r := base.copyOf(t) + rem := retirementOf(r) + reached := false + rem.Stop = func(p retention.Point) error { + if p.Phase == at.Phase && (at.Done == anyDone || p.Done == at.Done) { + reached = true + return errCrash + } + return nil + } + if res := rem.Pass(retention.Input{}); !reached || len(res.Errors) == 0 { + t.Fatalf("the retirement did not stop there: %+v", res) + } + before := treeOf(t, r.zone.SessionDir(session), filepath.Join(r.dir, "_conversations", session)) + r.collectAt(landedAt(10)) + r.derive() + r.parse() + if after := treeOf(t, r.zone.SessionDir(session), filepath.Join(r.dir, "_conversations", session)); strings.Join(after, "\n") != strings.Join(before, "\n") { + t.Fatalf("what is left of the stopped retirement changed:\nbefore:\n%s\nafter:\n%s", + strings.Join(before, "\n"), strings.Join(after, "\n")) + } + + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + rem2 := &retention.Remover{Zone: r.zone} + for i := 0; i < 2; i++ { + if res := rem2.Recover(); len(res.Errors) > 0 || len(res.Retired) > 0 { + t.Fatalf("a recovery after the finished one did something: %+v", res) + } + } + retiredWhole(t, r.zone, session, names(waiting)...) + sameRequests(t, r.dir, waiting) + + r.rcv.Reset() + r.push() + if got := len(r.rcv.MetricsRequests()); got != len(waiting) { + t.Fatalf("%d metrics requests were sent, want the %d that waited", got, len(waiting)) + } + noRecordOf(t, r.rcv, session) + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + }) + } +} + +// failingSync fails the nth sync of a directory, counting from one, and +// syncs every other one. +type failingSync struct { + dir string + nth int + seen int +} + +func (f *failingSync) sync(dir string) error { + if dir == f.dir { + f.seen++ + if f.seen == f.nth { + return errors.New("the disk did not take it") + } + } + return storage.SyncDir(dir) +} + +// TestRetirementStopsWhenItMayNotSurviveAPowerLoss. A retirement syncs the +// marker's directory, then the root, then the spool after each request it +// writes again and the root after them, and the spool after it takes away +// what the receiver has. A sync that fails stops the retirement there, in +// the pass and in the recovery after it, until a sync succeeds. Before the +// directories go, nothing is deleted; after them, the state keeps every line +// of the session. A power loss that took the marker back loses nothing, and +// one that brought a deleted request back sends nothing again. +func TestRetirementStopsWhenItMayNotSurviveAPowerLoss(t *testing.T) { + base, session, waiting := retiringRoot(t, "provider-bodies-long") + // nth counts the syncs of the directory in each attempt. A recovery + // after the directories went writes no request again, so the tidy's sync + // is its first of the spool. + for _, c := range []struct { + name string + dir func(root string) string + nth [2]int + deleted bool + }{ + {"the marker's directory", func(root string) string { return filepath.Join(root, storage.RetiredDir) }, [2]int{1, 1}, false}, + {"the root after the marker", func(root string) string { return root }, [2]int{1, 1}, false}, + {"the spool after a request", func(root string) string { return filepath.Join(root, storage.SpoolDir) }, [2]int{1, 1}, false}, + {"the root after the requests", func(root string) string { return root }, [2]int{2, 2}, false}, + {"the spool after the tidy", func(root string) string { return filepath.Join(root, storage.SpoolDir) }, [2]int{len(waiting) + 1, 1}, true}, + } { + t.Run(c.name, func(t *testing.T) { + r := base.copyOf(t) + answered := map[string][]byte{} + for _, name := range spoolOf(t, r.dir, session) { + if _, ok := waiting[name]; !ok { + data, err := os.ReadFile(filepath.Join(r.dir, storage.SpoolDir, name)) + if err != nil { + t.Fatal(err) + } + answered[name] = data + } + } + sessionDir := r.zone.SessionDir(session) + marker := r.zone.RetiredPath(session) + for attempt := 0; attempt < 2; attempt++ { + rem := retirementOf(r) + rem.SyncDir = (&failingSync{dir: c.dir(r.dir), nth: c.nth[attempt]}).sync + var res retention.Result + if attempt == 0 { + res = rem.Pass(retention.Input{}) + } else { + res = rem.Recover() + } + if len(res.Errors) == 0 { + t.Fatalf("attempt %d: a failed sync did not stop the retirement", attempt+1) + } + data, err := os.ReadFile(marker) + if err != nil { + t.Fatalf("the marker is not there, so a writer would not pass the session over: %v", err) + } + // Every attempt that deletes anything writes the marker + // again: a sync that succeeds after one that failed does + // not prove the first one's data reached the disk. + if !c.deleted { + if attempt == 0 { + if err := os.WriteFile(marker, []byte("data not saved"), 0o644); err != nil { + t.Fatal(err) + } + } else if len(data) != 0 { + t.Fatal("the retry did not write the marker again") + } + } + if present(sessionDir) == c.deleted { + t.Fatalf("attempt %d: the session's directory is there: %v; want deleted %v", attempt+1, present(sessionDir), c.deleted) + } + if c.deleted { + // Every line stays, the owner too, until the deletions + // are durable. + sent, err := otlpSent(r) + if err != nil { + t.Fatal(err) + } + if sent.Owner(session) == "" { + t.Fatal("the owner went before the tidy was durable") + } + for name := range answered { + if _, ok := sent.Digest(storage.SpoolDir + "/" + name); !ok { + t.Fatalf("the line of %s went before its deletion was durable", name) + } + } + } + } + if c.deleted { + // A power loss brings back a deleted request. It is not sent + // again, and the next recovery takes it away again. + for name, data := range answered { + if err := os.WriteFile(filepath.Join(r.dir, storage.SpoolDir, name), data, 0o444); err != nil { + t.Fatal(err) + } + } + r.rcv.Reset() + r.push() + if got := len(r.rcv.MetricsRequests()); got != len(waiting) { + t.Fatalf("%d metrics requests were sent, want only the %d that waited", got, len(waiting)) + } + } else { + // A power loss takes the marker back. Nothing was deleted, + // so nothing is lost, and the session is retired again. + if err := os.Remove(marker); err != nil { + t.Fatal(err) + } + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 || strings.Join(res.Retired, ",") != session { + t.Fatalf("the session was not retired again: %+v", res) + } + if _, err := os.Stat(marker); err != nil { + t.Fatalf("the marker was not written again: %v", err) + } + retiredWhole(t, r.zone, session, names(waiting)...) + return + } + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + }) + } +} + +// TestARetiredRequestIsWrittenAgainFromItsReceipt. A request the receiver +// does not have is written again from its receipt before the receipt goes. +// A failed writeback may have lost the request's data on disk. The test +// gives it different bytes and makes it read-only again. Retirement must +// restore the receipt's bytes and keep the request read-only. +// A retirement may also stop after deleting a request but before writing +// it again. Recovery must write it again from the receipt. +func TestARetiredRequestIsWrittenAgainFromItsReceipt(t *testing.T) { + base, session, waiting := retiringRoot(t, "provider-bodies-long") + t.Run("replaced", func(t *testing.T) { + r := base.copyOf(t) + for name := range waiting { + path := filepath.Join(r.dir, storage.SpoolDir, name) + if err := os.Chmod(path, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("data not saved"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.Chmod(path, storage.PermLanded); err != nil { + t.Fatal(err) + } + } + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + for name := range waiting { + after, err := os.Stat(filepath.Join(r.dir, storage.SpoolDir, name)) + if err != nil { + t.Fatal(err) + } + if after.Mode().Perm() != storage.PermLanded { + t.Fatalf("%s is %v, want read-only", name, after.Mode().Perm()) + } + } + sameRequests(t, r.dir, waiting) + }) + t.Run("deleted before its write", func(t *testing.T) { + r := base.copyOf(t) + rem := retirementOf(r) + rem.Stop = func(p retention.Point) error { + if p.Phase == retention.PhaseRetireMarked { + return errCrash + } + return nil + } + if res := rem.Pass(retention.Input{}); len(res.Errors) == 0 { + t.Fatal("the retirement did not stop") + } + for name := range waiting { + if err := storage.DeleteTree(filepath.Join(r.dir, storage.SpoolDir, name), nil); err != nil { + t.Fatal(err) + } + } + // One remover for every recovery, as a pipeline keeps one: the + // requests the recovery wrote again are waiting, so the retirement + // is not taken for finished. + rec := &retention.Remover{Zone: r.zone} + if res := rec.Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session, names(waiting)...) + sameRequests(t, r.dir, waiting) + r.push() + if res := rec.Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + }) +} + +// TestAHandOffReadsEveryReceipt. A receipt that is empty is a reservation a +// crash left, which recorded nothing, and the hand-off passes it over. A +// receipt that cannot be read stops the retirement before anything is +// deleted, since the request it makes could otherwise be lost. +func TestAHandOffReadsEveryReceipt(t *testing.T) { + base, session, waiting := retiringRoot(t, "provider-bodies-long") + receipt := func(r *sentRoot, name string) string { + return filepath.Join(metrics.ReceiptsDir(r.zone, session), name) + } + t.Run("a reservation", func(t *testing.T) { + r := base.copyOf(t) + if err := os.WriteFile(receipt(r, "999999.json"), nil, 0o444); err != nil { + t.Fatal(err) + } + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 || len(res.Retired) != 1 { + t.Fatalf("%+v", res) + } + retiredWhole(t, r.zone, session, names(waiting)...) + }) + t.Run("unreadable", func(t *testing.T) { + r := base.copyOf(t) + if err := os.WriteFile(receipt(r, "999999.json"), []byte(`{"schema":2,"request":"AA=="}`), 0o444); err != nil { + t.Fatal(err) + } + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) == 0 || len(res.Retired) != 0 { + t.Fatalf("a receipt that cannot be read did not stop the retirement: %+v", res) + } + if !present(r.zone.SessionDir(session)) { + t.Fatal("the session's directory went with a receipt that could not be read") + } + if left := spoolOf(t, r.dir, session); len(left) <= len(waiting) { + // The requests sent are still there: nothing was tidied. + t.Fatalf("the spool holds %v; the stopped retirement deleted from it", left) + } + }) +} + +// TestARetiredSessionIsNotGivenLevelOne. A retirement that stopped after its +// marker leaves the session's files, and a later configuration may give it +// level 1 alone. Level 1 passes a retired session over, so no skeleton is +// written into it: only its retirement goes on with it. +func TestARetiredSessionIsNotGivenLevelOne(t *testing.T) { + r := newSentRoot(t) + b := growthAt(t, r, 2, func(k int) { r.pipelineAt(landedAt(k)) }) + rem := retirementOf(r) + rem.Stop = func(p retention.Point) error { + if p.Phase == retention.PhaseRetireMarked { + return errCrash + } + return nil + } + if res := rem.Pass(retention.Input{}); len(res.Errors) == 0 { + t.Fatal("the retirement did not stop") + } + before := treeOf(t, r.zone.SessionDir(b.Session)) + levelOne := &retention.Remover{Zone: r.zone, MetadataOnlyAfter: time.Hour, Metrics: true, Now: later} + if res := levelOne.Pass(retention.Input{}); len(res.Errors) > 0 || res.Files != 0 { + t.Fatalf("level 1 over a retired session: %+v", res) + } + if after := treeOf(t, r.zone.SessionDir(b.Session)); strings.Join(after, "\n") != strings.Join(before, "\n") { + t.Fatalf("level 1 wrote into a retired session:\nbefore:\n%s\nafter:\n%s", strings.Join(before, "\n"), strings.Join(after, "\n")) + } +} + +// pushAs sends what is not sent yet under the service given, and returns +// what the pass did. +func (r *sentRoot) pushAs(service string) *otlp.Stats { + r.t.Helper() + client, err := r.rcv.Client(otlp.ProtocolGRPC) + if err != nil { + r.t.Fatal(err) + } + defer client.Close() + p := &otlp.Pusher{Zone: r.zone, Client: client, Endpoint: r.endpoint, Version: "test", + ServiceName: service, InstanceID: "removal-test", Layer: "AI_AGENT"} + st, err := p.Pass() + if err != nil { + r.t.Fatal(err) + } + return st +} + +// TestARetiredSessionsRequestsGoToItsOwner. The metrics requests of a +// retired session stay until they are sent, and go to its owner only. The +// owner stays in push.state while any of them waits. +// - Never sent before: no owner is kept, and the first send pins one. +// - Sent before: a pipeline that names another owner holds them; once it +// names the same one again, they go. +// - A request the receiver took and rejected part of is not sent again, +// and goes. +// - A send whose push.state save failed leaves the requests waiting, with +// their owner. They are sent again, at least once, and then go. +func TestARetiredSessionsRequestsGoToItsOwner(t *testing.T) { + t.Run("never sent", func(t *testing.T) { + r := newSentRoot(t) + b := growthAt(t, r, 2, func(k int) { + r.collectAt(landedAt(k)) + r.derive() + r.parse() + }) + requests := spoolOf(t, r.dir, b.Session) + if len(requests) < 2 { + t.Fatalf("%d requests; the test needs several", len(requests)) + } + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 || len(res.Retired) != 1 { + t.Fatalf("%+v", res) + } + retiredWhole(t, r.zone, b.Session, requests...) + if sent, err := otlpSent(r); err != nil || sent.Owner(b.Session) != "" { + t.Fatalf("a session never sent has an owner: %v", err) + } + r.rcv.Reset() + r.push() + if got := len(r.rcv.MetricsRequests()); got != len(requests) { + t.Fatalf("%d requests sent, want %d", got, len(requests)) + } + if sent, err := otlpSent(r); err != nil || sent.Owner(b.Session) == "" { + t.Fatalf("the first send pinned no owner: %v", err) + } + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, b.Session) + }) + t.Run("another owner", func(t *testing.T) { + r, session, waiting := retiringRoot(t, "provider-bodies-long") + sent, err := otlpSent(r) + if err != nil { + t.Fatal(err) + } + owner := sent.Owner(session) + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + r.rcv.Reset() + if st := r.pushAs("Another Service"); len(st.Held) != 1 || st.Held[0].Session != session { + t.Fatalf("another owner did not hold the session: %+v", st) + } + if got := len(r.rcv.MetricsRequests()); got != 0 { + t.Fatalf("%d requests went to another owner", got) + } + // One remover for every recovery, as a pipeline keeps one: a + // retirement with requests waiting is not taken for finished. + rec := &retention.Remover{Zone: r.zone} + if res := rec.Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session, names(waiting)...) + if sent, err := otlpSent(r); err != nil || sent.Owner(session) != owner { + t.Fatalf("the owner is %q, want %q kept: %v", sent.Owner(session), owner, err) + } + r.push() + if got := len(r.rcv.MetricsRequests()); got != len(waiting) { + t.Fatalf("%d requests sent under the owner, want %d", got, len(waiting)) + } + if res := rec.Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + }) + // A send can succeed while saving push.state fails. Recovery must keep + // the requests and their owner until a later send is saved. + t.Run("state save failed after send", func(t *testing.T) { + r, session, waiting := retiringRoot(t, "provider-bodies-long") + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session, names(waiting)...) + sent, err := otlpSent(r) + if err != nil { + t.Fatal(err) + } + owner := sent.Owner(session) + if owner == "" { + t.Fatal("the retirement left no owner for the waiting requests") + } + state := filepath.Join(r.dir, otlp.StateFile) + saved, err := os.ReadFile(state) + if err != nil { + t.Fatal(err) + } + r.rcv.Reset() + r.push() + if got := len(r.rcv.MetricsRequests()); got != len(waiting) { + t.Fatalf("%d requests reached the receiver, want %d", got, len(waiting)) + } + if err := os.WriteFile(state, saved, storage.PermState); err != nil { + t.Fatal(err) + } + rec := &retention.Remover{Zone: r.zone} + if res := rec.Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session, names(waiting)...) + if sent, err := otlpSent(r); err != nil { + t.Fatal(err) + } else if sent.Owner(session) != owner { + t.Fatalf("the owner is %q, want %q kept after the failed save", sent.Owner(session), owner) + } + r.push() + if got := len(r.rcv.MetricsRequests()); got != 2*len(waiting) { + t.Fatalf("%d requests reached the receiver, want %d after sending them again", got, 2*len(waiting)) + } + if res := rec.Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + }) + t.Run("rejected in part", func(t *testing.T) { + r, session, waiting := retiringRoot(t, "provider-bodies-long") + if res := retirementOf(r).Pass(retention.Input{}); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + r.rcv.Reject(1) + st := r.pushAs("Removal Test") + r.rcv.Reject(0) + if st.Rejected == 0 { + t.Fatalf("the receiver rejected nothing: %+v", st) + } + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + r.rcv.Reset() + r.push() + if got := len(r.rcv.MetricsRequests()); got != 0 { + t.Fatalf("%d requests were sent again, of %d", got, len(waiting)) + } + }) +} + +// TestThePageDuringARetirement. The page drew a conversation and keeps it. +// While its retirement is unfinished, the page lists and draws it as long as +// its chain is there, and once the chain is gone it does neither. No read of +// the page, by its kept paths or by new ones, makes anything of the session +// again, at any point of the retirement and after it. +func TestThePageDuringARetirement(t *testing.T) { + base := newSentRoot(t) + b := growthAt(t, base, 2, func(k int) { base.pipelineAt(landedAt(k)) }) + id := b.Session + get := func(h http.Handler, path string) (int, string) { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil)) + return rec.Code, rec.Body.String() + } + paths := []string{"/api/c/" + id + "/view", "/api/c/" + id + "/record/1/1", "/api/c/" + id + "/files?seq=1"} + for _, at := range []string{retention.PhaseRetireMarked, retention.PhaseRetireSession + " renamed", + retention.PhaseRetireSession + " removed", retention.PhaseRetireChain + " renamed", ""} { + name := at + if name == "" { + name = "finished" + } + t.Run(name, func(t *testing.T) { + r := base.copyOf(t) + h := view.New(r.zone, nil).Handler() + for _, path := range paths[:2] { + if code, body := get(h, path); code != http.StatusOK { + t.Fatalf("%s before the retirement: %d %s", path, code, body) + } + } + get(h, paths[2]) + rem := retirementOf(r) + rem.Stop = func(p retention.Point) error { + if p.Phase == at { + return errCrash + } + return nil + } + rem.Pass(retention.Input{}) + sessionDir, chainDir := r.zone.SessionDir(id), filepath.Join(r.dir, "_conversations", id) + before := treeOf(t, sessionDir, chainDir, filepath.Join(r.dir, storage.RemovedDir)) + chain := present(chainDir) + _, list := get(h, "/api/conversations") + if listed := strings.Contains(list, `"id":"`+id+`"`); listed != chain { + t.Fatalf("the list names it: %v; its chain is there: %v", listed, chain) + } + if code, body := get(h, paths[0]); (code == http.StatusOK) != chain { + t.Fatalf("the conversation answers %d with its chain there: %v: %s", code, chain, body) + } + for _, path := range paths[1:] { + get(h, path) + } + fresh := view.New(r.zone, nil).Handler() + for _, path := range paths { + get(fresh, path) + } + if after := treeOf(t, sessionDir, chainDir, filepath.Join(r.dir, storage.RemovedDir)); strings.Join(after, "\n") != strings.Join(before, "\n") { + t.Fatalf("a read of the page changed the root:\nbefore:\n%s\nafter:\n%s", strings.Join(before, "\n"), strings.Join(after, "\n")) + } + }) + } +} + +// copyDir copies a directory tree, modes kept. +func copyDir(t *testing.T, from, to string) { + t.Helper() + err := filepath.WalkDir(from, func(p string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + rel, err := filepath.Rel(from, p) + if err != nil { + return err + } + info, err := d.Info() + if err != nil { + return err + } + dst := filepath.Join(to, rel) + if d.IsDir() { + return os.MkdirAll(dst, info.Mode().Perm()|0o700) + } + data, err := os.ReadFile(p) + if err != nil { + return err + } + return os.WriteFile(dst, data, info.Mode().Perm()) + }) + if err != nil { + t.Fatal(err) + } +} + +// TestRetirementForgetsNothingBeforeItsDirectoriesAreDurable. Each rename +// checks both directory syncs before deleting what it moved. The retirement +// checks the directories again before the state forgets anything. A failure +// at either check leaves push.state unchanged, including the owner. A retry +// whose final directory check fails must keep that state too. A power loss +// that brings back the directories, receipts and answered requests must not +// send those requests again. Recovery finishes the retirement. +func TestRetirementForgetsNothingBeforeItsDirectoriesAreDurable(t *testing.T) { + base, session, waiting := retiringRoot(t, "provider-bodies-long") + // The first attempt syncs the root for the marker, the requests, the + // session rename and the final check, in that order. It syncs the + // conversations directory for the chain rename and the final check. + // It syncs _removed for each rename, then for the final check. + // Each retry fails the final check of the same directory. A failed + // rename sync leaves directories to remove, so the retry writes the + // marker again and may still rename the chain. A failed final check + // leaves no directories or answered requests, so only the checks remain. + const ( + handed = retention.PhaseRetireHanded + sessionGone = retention.PhaseRetireSession + " removed" + tidied = retention.PhaseRetireTidied + ) + for _, c := range []struct { + name string + dir func(root string) string + nth [2]int + after [2]string + }{ + {"the root after the session rename", func(root string) string { return root }, [2]int{3, 2}, [2]string{handed, tidied}}, + {"the removed directory after the session rename", func(root string) string { return filepath.Join(root, storage.RemovedDir) }, [2]int{1, 2}, [2]string{handed, tidied}}, + {"the conversations after the chain rename", func(root string) string { return filepath.Join(root, "_conversations") }, [2]int{1, 1}, [2]string{sessionGone, tidied}}, + {"the removed directory after the chain rename", func(root string) string { return filepath.Join(root, storage.RemovedDir) }, [2]int{2, 1}, [2]string{sessionGone, tidied}}, + {"the root before forgetting", func(root string) string { return root }, [2]int{4, 1}, [2]string{tidied, ""}}, + {"the conversations before forgetting", func(root string) string { return filepath.Join(root, "_conversations") }, [2]int{2, 1}, [2]string{tidied, ""}}, + {"the removed directory before forgetting", func(root string) string { return filepath.Join(root, storage.RemovedDir) }, [2]int{3, 1}, [2]string{tidied, ""}}, + } { + t.Run(c.name, func(t *testing.T) { + r := base.copyOf(t) + stateBefore, err := os.ReadFile(filepath.Join(r.dir, otlp.StateFile)) + if err != nil { + t.Fatal(err) + } + answered := map[string][]byte{} + for _, name := range spoolOf(t, r.dir, session) { + if _, ok := waiting[name]; !ok { + data, err := os.ReadFile(filepath.Join(r.dir, storage.SpoolDir, name)) + if err != nil { + t.Fatal(err) + } + answered[name] = data + } + } + for attempt := 0; attempt < 2; attempt++ { + rem := retirementOf(r) + failure := &failingSync{dir: c.dir(r.dir), nth: c.nth[attempt]} + rem.SyncDir = failure.sync + last := "" + rem.Stop = func(p retention.Point) error { + last = p.Phase + return nil + } + var res retention.Result + if attempt == 0 { + res = rem.Pass(retention.Input{}) + } else { + res = rem.Recover() + } + if failure.seen != failure.nth || len(res.Errors) != 1 || !strings.Contains(res.Errors[0].Error(), "the disk did not take it") { + t.Fatalf("attempt %d: sync %d of %d did not stop the retirement: %v", attempt+1, failure.seen, failure.nth, res.Errors) + } + if last != c.after[attempt] { + t.Fatalf("attempt %d: the sync failed after %q, want after %q", attempt+1, last, c.after[attempt]) + } + if present(r.zone.SessionDir(session)) { + t.Fatalf("attempt %d: the session's directory is still there", attempt+1) + } + stateAfter, err := os.ReadFile(filepath.Join(r.dir, otlp.StateFile)) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(stateAfter, stateBefore) { + t.Fatalf("attempt %d: push.state changed before the directories were durable", attempt+1) + } + } + restoreAfterPowerLoss(t, base, r, session, answered) + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session, names(waiting)...) + r.rcv.Reset() + r.push() + if got := len(r.rcv.MetricsRequests()); got != len(waiting) { + t.Fatalf("%d metrics requests were sent, want only the %d that waited", got, len(waiting)) + } + noRecordOf(t, r.rcv, session) + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session) + }) + } +} + +// restoreAfterPowerLoss puts back what a power loss would bring back when +// the renames and deletions of a retirement did not reach the disk: the +// session's directory and its conversation's, as they were before, and the +// requests the receiver has. A lost rename leaves only the old name, so its +// destination under _removed must go before the old directory is restored. +// Anything the retirement did not reach is replaced by the same contents. +func restoreAfterPowerLoss(t *testing.T, base, r *sentRoot, session string, answered map[string][]byte) { + t.Helper() + for _, name := range []string{session, session + ".chain"} { + if err := storage.DeleteTree(filepath.Join(r.dir, storage.RemovedDir, name), nil); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{session, filepath.Join("_conversations", session)} { + if err := storage.DeleteTree(filepath.Join(r.dir, dir), nil); err != nil { + t.Fatal(err) + } + copyDir(t, filepath.Join(base.dir, dir), filepath.Join(r.dir, dir)) + } + for name, data := range answered { + path := filepath.Join(r.dir, storage.SpoolDir, name) + if err := storage.DeleteTree(path, nil); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, data, 0o444); err != nil { + t.Fatal(err) + } + } +} + +// TestARetirementStoppedAfterItsDirectoriesIsCounted. A retirement whose +// directories are gone and which then stops, before the spool tidy is done, +// still names the session as retired. No later attempt finds a directory to +// say so, and the pipeline counts the session and drops it from the page's +// memory by that name. +func TestARetirementStoppedAfterItsDirectoriesIsCounted(t *testing.T) { + r, session, waiting := retiringRoot(t, "provider-bodies-long") + rem := retirementOf(r) + rem.Stop = func(p retention.Point) error { + if p.Phase == retention.PhaseRetireTidied { + return errCrash + } + return nil + } + res := rem.Pass(retention.Input{}) + if len(res.Errors) == 0 { + t.Fatal("the retirement did not stop") + } + if strings.Join(res.Retired, ",") != session { + t.Fatalf("the stopped retirement names %v as retired, want %s", res.Retired, session) + } + if res := (&retention.Remover{Zone: r.zone}).Recover(); len(res.Errors) > 0 { + t.Fatal(res.Errors) + } + retiredWhole(t, r.zone, session, names(waiting)...) +} diff --git a/tests/chain/retired_test.go b/tests/chain/retired_test.go index 4eb8173..07a7880 100644 --- a/tests/chain/retired_test.go +++ b/tests/chain/retired_test.go @@ -38,6 +38,9 @@ import ( // treeOf is every file under the directories given, with its mode, size and // time, so a write of any kind shows. A directory that is not there says so. +// A directory is listed by its name and mode only: Windows updates a +// directory's time some moments after a file in it is written, and an entry +// added or removed shows by its own name. func treeOf(t *testing.T, dirs ...string) []string { t.Helper() var out []string @@ -51,6 +54,10 @@ func treeOf(t *testing.T, dirs ...string) []string { return err } rel, _ := filepath.Rel(dir, path) + if d.IsDir() { + out = append(out, fmt.Sprintf("%s/%s %s", filepath.Base(dir), filepath.ToSlash(rel), info.Mode())) + return nil + } out = append(out, fmt.Sprintf("%s/%s %s %d %s", filepath.Base(dir), filepath.ToSlash(rel), info.Mode(), info.Size(), info.ModTime())) return nil diff --git a/tests/langchain_retire_test.go b/tests/langchain_retire_test.go new file mode 100644 index 0000000..74aa1c4 --- /dev/null +++ b/tests/langchain_retire_test.go @@ -0,0 +1,92 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package tests_test + +import ( + "bytes" + "os" + "path/filepath" + "testing" + "time" + + "github.com/apache/skywalking-ai-sessionizer/internal/adapters/langsmith" + "github.com/apache/skywalking-ai-sessionizer/internal/parse" + "github.com/apache/skywalking-ai-sessionizer/internal/retention" + "github.com/apache/skywalking-ai-sessionizer/internal/storage" +) + +// TestALangChainConversationIsRetiredByTheSameLevels. A LangChain +// conversation has no source to outlive, so asz alone gives it a lifetime, +// by the same two levels as every runtime. Once it is idle for retire_after, +// everything of it goes, and the runs remembered in open.json are kept as +// they were: an update carries nothing to find its conversation by. A later +// turn of its thread is not collected, and nothing of it is made again. +func TestALangChainConversationIsRetiredByTheSameLevels(t *testing.T) { + zone, post, collect := receiveInto(t) + owner := `"session_name":"asz","extra":{"metadata":{"thread_id":"t-retired"}}` + turn := func(id, question, at string) string { + return `{"post":[{"id":"` + id + `","trace_id":"` + id + `",` + + `"dotted_order":"20260920T10000` + at + `000000Z` + id + `",` + + `"run_type":"chain","name":"agent","start_time":"2026-09-20T10:00:0` + at + `Z",` + + `"end_time":"2026-09-20T10:00:0` + at + `Z",` + owner + `,` + + `"inputs":{"messages":[{"role":"user","content":"` + question + `"}]},` + + `"outputs":{"messages":[{"role":"assistant","content":"ok"}]}}]}` + } + post(turn("52525252-5252-7252-8252-525252525251", "first question", "1")) + session := collect().Sessions[0] + for { + round, err := parse.Session(zone, parse.Options{Conversation: session, Session: session}) + if err != nil { + t.Fatal(err) + } + if !round.Changed() || !round.More { + break + } + } + pending := filepath.Join(zone.Root(), langsmith.InboxDir, "open.json") + remembered, err := os.ReadFile(pending) + if err != nil { + t.Fatal(err) + } + + rem := &retention.Remover{Zone: zone, MetadataOnlyAfter: time.Hour, RetireAfter: time.Hour, + Now: func() time.Time { return time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC) }} + if res := rem.Pass(retention.Input{}); len(res.Errors) > 0 || len(res.Retired) != 1 || res.Retired[0] != session { + t.Fatalf("the conversation was not retired: %+v", res) + } + if retired, err := storage.Retired(zone, session); err != nil || !retired { + t.Fatalf("no marker: %v", err) + } + for _, dir := range []string{zone.SessionDir(session), filepath.Join(zone.Root(), "_conversations", session)} { + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("%s is left: %v", dir, err) + } + } + if after, err := os.ReadFile(pending); err != nil || !bytes.Equal(after, remembered) { + t.Fatalf("the retirement changed what open.json remembers: %v", err) + } + + post(turn("52525252-5252-7252-8252-525252525252", "second question", "5")) + landed := collect() + if len(landed.Sessions) != 0 || len(landed.Retired) != 1 || landed.Retired[0] != session { + t.Fatalf("the later turn of a retired conversation: %+v", landed) + } + if _, err := os.Stat(zone.SessionDir(session)); !os.IsNotExist(err) { + t.Fatalf("the retired conversation's directory was made again: %v", err) + } +}