Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions cmd/ax-server/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
60 changes: 60 additions & 0 deletions internal/controller/orphan_check.go
Original file line number Diff line number Diff line change
@@ -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
}
101 changes: 101 additions & 0 deletions internal/controller/orphan_check_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
})
}
46 changes: 46 additions & 0 deletions internal/controller/orphans.go
Original file line number Diff line number Diff line change
@@ -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
}
85 changes: 85 additions & 0 deletions internal/controller/orphans_test.go
Original file line number Diff line number Diff line change
@@ -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])
}
}
})
}
}
50 changes: 50 additions & 0 deletions internal/substrate/list_actors.go
Original file line number Diff line number Diff line change
@@ -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
}
}
}
Loading
Loading