[rakyll]

Replace the Redis store with PostgreSQL, and stop storing the parts of a
task's status that belong to its actor on Agent Substrate.

Storage:

• internal/store/postgres holds resources as protojson JSONB documents.
  Open creates any missing tables. Locks are leases in a locks table,
  renewed while held.
• Watches and lock waiters poll. There is no LISTEN/NOTIFY, so any
  connection pooler works. WatchTask is removed from store.Store.
• The memory store is now test-only and no longer tracks save order.
  Newest-first ordering is covered by a Postgres-specific test.

Task status:

• The store keeps only what AX owns: spec, metadata, the Pending, Failed
  and Terminating phases, and WorkspaceReady. The actor, worker IP, live
  phase and Ready condition are read from Substrate on every get, list
  and watch.
• Only a few helpers in server.go (getTask, listTasks, saveTask,
  updateTask, deleteTask) touch task storage, and a test enforces it.
• Phases, condition types and condition reasons are now constants in
  pkg/apis/v1alpha1. Ready is derived from one rule in the server
  instead of being written in several places.
• Ready reasons visible to clients change: TaskSuspended, ActorResuming,
  ActorSuspending, ActorCrashed, ActorDeleting and UnknownActorState
  become the phase name (Suspended, Resuming, Suspending, Failed,
  Terminating, Unknown).

This work revealed major problems in the server, storage and controller
implementations. Among them:

• Task status was copied from Substrate into storage and could drift
  from the actor it described.
• The controller wrote Ready conditions and phases that the server then
  discarded, or that disagreed with what clients saw.
• Handlers pass the desired action to the reconciler by overwriting
  status.phase, mixing intent with observed state.
• Phase and reason strings were spread across packages as literals. The
  server and CLI checked for a "Completed" phase that nothing produced.
• The shared store tests encoded one backend's list ordering as a
  contract.

This change fixes some of these, but the server, store and controller
need further stabilization. @rakyll is taking ownership of that work.

[rich7420]

@rakyll thanks for the patch! left some comments below

[rich7420]

Fresh timestamps trigger MODIFIED on unchanged reads. Could we omit unknown Ready.updateTime from get/list/watch responses?

Suggested change

	// setReady sets the task's Ready condition. Its update time changes only when
	// the condition does, so that repeated reads report a stable timestamp.
	func setReady(task *v1alpha1.Task, condStatus, reason, message string) {
	for _, c := range task.Status.Conditions {
	if c.GetType() != v1alpha1.ConditionReady {
	continue
	}
	if c.Status != condStatus || c.Reason != reason || c.Message != message {
	c.Status, c.Reason, c.Message = condStatus, reason, message
	c.UpdateTime = timestamppb.Now()
	}
	return
	}
	task.Status.Conditions = append(task.Status.Conditions, &v1alpha1.Condition{
	Type: v1alpha1.ConditionReady,
	Status: condStatus,
	Reason: reason,
	Message: message,
	UpdateTime: timestamppb.Now(),
	})
	}
	// setReady sets the derived Ready condition. No condition update time is
	// available from its source, so UpdateTime is left unset.
	func setReady(task *v1alpha1.Task, condStatus, reason, message string) {
	for _, c := range task.Status.Conditions {
	if c.GetType() != v1alpha1.ConditionReady {
	continue
	}
	c.Status, c.Reason, c.Message = condStatus, reason, message
	c.UpdateTime = nil
	return
	}
	task.Status.Conditions = append(task.Status.Conditions, &v1alpha1.Condition{
	Type: v1alpha1.ConditionReady,
	Status: condStatus,
	Reason: reason,
	Message: message,
	})
	}

[rich7420]

With the setReady change, could we assert that derived Ready.updateTime is unset?

Suggested change

	t.Run("stable update time", func(t *testing.T) {
	task := storedTask("suspended", "")
	setActorStatus(task, testActor("suspended", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ""), nil)
	first := readyCondition(task).GetUpdateTime()
	setActorStatus(task, testActor("suspended", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ""), nil)
	if got := readyCondition(task).GetUpdateTime(); got != first {
	t.Errorf("Ready update time changed on an unchanged read")
	}
	})
	t.Run("derived Ready has no update time", func(t *testing.T) {
	task := storedTask("suspended", "")
	setActorStatus(task, testActor("suspended", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ""), nil)
	if got := readyCondition(task).GetUpdateTime(); got != nil {
	t.Errorf("derived Ready update time = %v, want unset", got)
	}
	})

[rich7420]

Redacted() misses query and keyword-DSN passwords. Could we strip URL parameters and hide non-URL DSNs?

Suggested change

	// redactURL hides any password in a store URL so that it can be logged.
	func redactURL(raw string) string {
	u, err := url.Parse(raw)
	if err != nil {
	return "<unparseable>"
	}
	return u.Redacted()
	}
	// redactURL strips URL parameters and masks the userinfo password for logging.
	// Other connection string formats are hidden entirely.
	func redactURL(raw string) string {
	u, err := url.Parse(raw)
	if err != nil {
	return "<unparseable>"
	}
	if (u.Scheme != "postgres" && u.Scheme != "postgresql") || u.Host == "" {
	return "<redacted>"
	}
	u.RawQuery, u.Fragment, u.RawFragment = "", "", ""
	u.ForceQuery = false
	return u.Redacted()
	}

[rich7420]

Could we fetch only this page’s actors? This avoids scanning the whole atespace, at the cost of one sequential lookup per task.

Suggested change

	// listTasks returns a page of tasks with their Substrate-owned status filled
	// in, using one ListActors call per atespace.
	func (s *Server) listTasks(ctx context.Context, atespace string, limit, offset int64) ([]*v1alpha1.Task, error) {
	tasks, err := s.store.ListTasks(ctx, atespace, limit, offset)
	if err != nil || s.substrate == nil {
	return tasks, err
	}
	byAtespace := make(map[string][]*v1alpha1.Task)
	for _, t := range tasks {
	ns := t.GetMetadata().GetAtespace()
	byAtespace[ns] = append(byAtespace[ns], t)
	}
	for ns, group := range byAtespace {
	actors, err := s.substrate.ListActors(ctx, ns)
	byName := make(map[string]*ateapipb.Actor, len(actors))
	for _, a := range actors {
	byName[a.GetMetadata().GetName()] = a
	}
	for _, t := range group {
	setActorStatus(t, byName[t.GetMetadata().GetName()], err)
	}
	}
	return tasks, nil
	}
	// listTasks returns a page of tasks with their Substrate-owned status filled in.
	func (s *Server) listTasks(ctx context.Context, atespace string, limit, offset int64) ([]*v1alpha1.Task, error) {
	tasks, err := s.store.ListTasks(ctx, atespace, limit, offset)
	if err != nil || s.substrate == nil {
	return tasks, err
	}
	for _, task := range tasks {
	s.applyActor(ctx, task)
	}
	return tasks, nil
	}