diff --git a/cmd/ax-server/main.go b/cmd/ax-server/main.go index 967b00b1..d97d12fd 100644 --- a/cmd/ax-server/main.go +++ b/cmd/ax-server/main.go @@ -128,6 +128,15 @@ func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer cancel() + // Log-only check, run before the server starts serving requests. + if subClient != nil { + checkCtx, cancelCheck := context.WithTimeout(ctx, 30*time.Second) + if err := controller.LogOrphanActors(checkCtx, subClient, rStore); err != nil { + slog.Warn("orphan check failed", "error", err) + } + cancelCheck() + } + go func() { if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { slog.Error("HTTP server failed", "error", err) diff --git a/internal/controller/orphan_check.go b/internal/controller/orphan_check.go new file mode 100644 index 00000000..4f41e5fc --- /dev/null +++ b/internal/controller/orphan_check.go @@ -0,0 +1,60 @@ +// Copyright 2026 Google LLC +// +// Licensed 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 controller + +import ( + "context" + "log/slog" + "math" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/ax/internal/store" +) + +// ActorLister lists every Substrate actor. *substrate.Client satisfies it. +type ActorLister interface { + ListAllActors(ctx context.Context) ([]*ateapipb.Actor, error) +} + +// LogOrphanActors logs every Substrate actor that has no matching Task record. +// It only logs; it never deletes anything. +func LogOrphanActors(ctx context.Context, client ActorLister, s store.Store) error { + // Actors are listed before tasks on purpose. A task saved after the actor + // list cannot cause a false orphan, because its actor is not in that list. + // A task deleted after the actor list can still cause one false warning. + // That is acceptable here because this check only logs. + actors, err := client.ListAllActors(ctx) + if err != nil { + return err + } + + // Read every task in one call. Paging by offset is unsafe here: a task + // deleted between pages shifts the list, so the scan can skip a task and + // log a live actor as an orphan. An empty atespace lists all atespaces. + const allTasks = math.MaxInt32 + tasks, err := s.ListTasks(ctx, "", allTasks, 0) + if err != nil { + return err + } + + orphans := FindOrphanActors(actors, tasks) + for _, a := range orphans { + slog.Warn("orphan actor has no task record", + "atespace", a.GetMetadata().GetAtespace(), + "actor", a.GetMetadata().GetName()) + } + slog.Info("orphan check finished", "actors", len(actors), "tasks", len(tasks), "orphans", len(orphans)) + return nil +} diff --git a/internal/controller/orphan_check_test.go b/internal/controller/orphan_check_test.go new file mode 100644 index 00000000..6cb5e10b --- /dev/null +++ b/internal/controller/orphan_check_test.go @@ -0,0 +1,101 @@ +// Copyright 2026 Google LLC +// +// Licensed 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 controller + +import ( + "context" + "errors" + "testing" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/ax/internal/store" + "github.com/google/ax/pkg/apis/v1alpha1" +) + +type fakeLister struct { + actors []*ateapipb.Actor + err error +} + +func (f fakeLister) ListAllActors(ctx context.Context) ([]*ateapipb.Actor, error) { + return f.actors, f.err +} + +// fakeStore embeds the interface so only ListTasks needs to be implemented. +type fakeStore struct { + store.Store + tasks []*v1alpha1.Task + err error + calls *int // counts ListTasks calls, when non-nil +} + +func (f fakeStore) ListTasks(ctx context.Context, atespace string, limit, offset int64) ([]*v1alpha1.Task, error) { + if f.calls != nil { + *f.calls++ + } + if f.err != nil { + return nil, f.err + } + if offset >= int64(len(f.tasks)) { + return nil, nil + } + end := min(offset+limit, int64(len(f.tasks))) + return f.tasks[offset:end], nil +} + +func TestLogOrphanActors(t *testing.T) { + ctx := context.Background() + + t.Run("runs without error when there are orphans", func(t *testing.T) { + lister := fakeLister{actors: []*ateapipb.Actor{actor("default", "gone")}} + s := fakeStore{tasks: []*v1alpha1.Task{task("default", "alive")}} + if err := LogOrphanActors(ctx, lister, s); err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + + t.Run("returns the actor listing error", func(t *testing.T) { + want := errors.New("substrate down") + lister := fakeLister{err: want} + if err := LogOrphanActors(ctx, lister, fakeStore{}); !errors.Is(err, want) { + t.Fatalf("got %v, want %v", err, want) + } + }) + + t.Run("returns the task listing error", func(t *testing.T) { + want := errors.New("redis down") + lister := fakeLister{actors: []*ateapipb.Actor{actor("default", "a")}} + if err := LogOrphanActors(ctx, lister, fakeStore{err: want}); !errors.Is(err, want) { + t.Fatalf("got %v, want %v", err, want) + } + }) + + t.Run("reads every task in one call", func(t *testing.T) { + // 1,001 tasks: a paged read would need two calls; this one needs one. + var tasks []*v1alpha1.Task + for i := 0; i < 1001; i++ { + tasks = append(tasks, task("default", "t")) + } + calls := 0 + s := fakeStore{tasks: tasks, calls: &calls} + if err := LogOrphanActors(ctx, fakeLister{}, s); err != nil { + t.Fatalf("unexpected error: %v", err) + } + // All tasks come back in a single read, so there is no paging to skip. + if calls != 1 { + t.Fatalf("ListTasks called %d times, want 1", calls) + } + }) +} diff --git a/internal/controller/orphans.go b/internal/controller/orphans.go new file mode 100644 index 00000000..e0d158e8 --- /dev/null +++ b/internal/controller/orphans.go @@ -0,0 +1,46 @@ +// Copyright 2026 Google LLC +// +// Licensed 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 controller + +import ( + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/ax/pkg/apis/v1alpha1" +) + +// taskKey identifies a task or actor by atespace and name. A struct key avoids +// collisions that a joined string would have when a name contains a separator. +type taskKey struct { + atespace, name string +} + +// FindOrphanActors returns the actors that have no matching task. +// An actor matches a task when both its name and its atespace equal the task's. +func FindOrphanActors(actors []*ateapipb.Actor, tasks []*v1alpha1.Task) []*ateapipb.Actor { + // Step 1: record each task's key so lookups are fast + taskKeys := map[taskKey]bool{} + for _, task := range tasks { + taskKeys[taskKey{task.GetMetadata().GetAtespace(), task.GetMetadata().GetName()}] = true + } + + // Step 2: keep each actor whose key is not in that set + var orphans []*ateapipb.Actor + for _, actor := range actors { + key := taskKey{actor.GetMetadata().GetAtespace(), actor.GetMetadata().GetName()} + if !taskKeys[key] { + orphans = append(orphans, actor) + } + } + return orphans +} diff --git a/internal/controller/orphans_test.go b/internal/controller/orphans_test.go new file mode 100644 index 00000000..01be8f1d --- /dev/null +++ b/internal/controller/orphans_test.go @@ -0,0 +1,85 @@ +// Copyright 2026 Google LLC +// +// Licensed 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 controller + +import ( + "testing" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/ax/pkg/apis/v1alpha1" +) + +func actor(atespace, name string) *ateapipb.Actor { + return &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: name}} +} + +func task(atespace, name string) *v1alpha1.Task { + return &v1alpha1.Task{Metadata: &v1alpha1.ObjectMeta{Atespace: atespace, Name: name}} +} + +func TestFindOrphanActors(t *testing.T) { + tests := []struct { + name string + actors []*ateapipb.Actor + tasks []*v1alpha1.Task + want []string // "atespace/name" of expected orphans + }{ + { + name: "no actors", + actors: nil, + tasks: []*v1alpha1.Task{task("default", "a")}, + want: nil, + }, + { + name: "matched pair is not an orphan", + actors: []*ateapipb.Actor{actor("default", "a")}, + tasks: []*v1alpha1.Task{task("default", "a")}, + want: nil, + }, + { + name: "actor with no task is an orphan", + actors: []*ateapipb.Actor{actor("default", "a"), actor("default", "b")}, + tasks: []*v1alpha1.Task{task("default", "a")}, + want: []string{"default/b"}, + }, + { + name: "same name in another atespace is an orphan", + actors: []*ateapipb.Actor{actor("team-x", "a")}, + tasks: []*v1alpha1.Task{task("default", "a")}, + want: []string{"team-x/a"}, + }, + { + name: "no tasks means every actor is an orphan", + actors: []*ateapipb.Actor{actor("default", "a")}, + tasks: nil, + want: []string{"default/a"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := FindOrphanActors(tc.actors, tc.tasks) + if len(got) != len(tc.want) { + t.Fatalf("got %d orphans, want %d", len(got), len(tc.want)) + } + for i, a := range got { + key := a.GetMetadata().GetAtespace() + "/" + a.GetMetadata().GetName() + if key != tc.want[i] { + t.Errorf("orphan %d = %q, want %q", i, key, tc.want[i]) + } + } + }) + } +} diff --git a/internal/substrate/list_actors.go b/internal/substrate/list_actors.go new file mode 100644 index 00000000..fa267bc5 --- /dev/null +++ b/internal/substrate/list_actors.go @@ -0,0 +1,50 @@ +// Copyright 2026 Google LLC +// +// Licensed 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 substrate + +import ( + "context" + "fmt" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" +) + +// ListAllActors returns every actor across all atespaces, following pagination. +func (c *Client) ListAllActors(ctx context.Context) ([]*ateapipb.Actor, error) { + var all []*ateapipb.Actor + pageToken := "" + for { + var resp *ateapipb.ListActorsResponse + err := retryOnCodes(ctx, func() error { + var err error + resp, err = c.control.ListActors(ctx, &ateapipb.ListActorsRequest{ + Options: &ateapipb.ListOptions{ + PageSize: 1000, + PageToken: pageToken, + }, + }) + return err + }, codes.Unavailable) + if err != nil { + return nil, fmt.Errorf("listing actors: %w", err) + } + all = append(all, resp.GetActors()...) + pageToken = resp.GetNextPageToken() + if pageToken == "" { + return all, nil + } + } +} diff --git a/internal/substrate/list_actors_test.go b/internal/substrate/list_actors_test.go new file mode 100644 index 00000000..f82953f1 --- /dev/null +++ b/internal/substrate/list_actors_test.go @@ -0,0 +1,87 @@ +// Copyright 2026 Google LLC +// +// Licensed 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 substrate + +import ( + "context" + "errors" + "testing" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc" +) + +// fakeControl overrides only ListActors; the embedded interface is nil, so any +// other call would panic and show up in the test. +type fakeControl struct { + ateapipb.ControlClient + pages []*ateapipb.ListActorsResponse + tokens []string + err error // returned on every call when set + errOnSecond error // returned only on the second call when set +} + +func (f *fakeControl) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest, _ ...grpc.CallOption) (*ateapipb.ListActorsResponse, error) { + f.tokens = append(f.tokens, req.GetOptions().GetPageToken()) + if f.err != nil { + return nil, f.err + } + if len(f.tokens) == 2 && f.errOnSecond != nil { + return nil, f.errOnSecond + } + page := f.pages[0] + f.pages = f.pages[1:] + return page, nil +} + +func TestListAllActorsFollowsPages(t *testing.T) { + fc := &fakeControl{pages: []*ateapipb.ListActorsResponse{ + {Actors: []*ateapipb.Actor{{Metadata: &ateapipb.ResourceMetadata{Name: "a"}}}, NextPageToken: "next"}, + {Actors: []*ateapipb.Actor{{Metadata: &ateapipb.ResourceMetadata{Name: "b"}}}}, + }} + c := &Client{control: fc} + + got, err := c.ListAllActors(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 2 || got[0].GetMetadata().GetName() != "a" || got[1].GetMetadata().GetName() != "b" { + t.Fatalf("got %d actors, want a then b", len(got)) + } + if len(fc.tokens) != 2 || fc.tokens[0] != "" || fc.tokens[1] != "next" { + t.Fatalf("page tokens = %q, want [\"\" \"next\"]", fc.tokens) + } +} + +func TestListAllActorsReturnsError(t *testing.T) { + want := errors.New("boom") + c := &Client{control: &fakeControl{err: want}} + if _, err := c.ListAllActors(context.Background()); err == nil { + t.Fatal("expected an error") + } +} + +func TestListAllActorsReturnsErrorOnSecondPage(t *testing.T) { + fc := &fakeControl{ + pages: []*ateapipb.ListActorsResponse{ + {Actors: []*ateapipb.Actor{{Metadata: &ateapipb.ResourceMetadata{Name: "a"}}}, NextPageToken: "next"}, + }, + errOnSecond: errors.New("second page failed"), + } + c := &Client{control: fc} + if _, err := c.ListAllActors(context.Background()); err == nil { + t.Fatal("expected an error from the second page, got nil") + } +}