feat: unify interactive operation coordination

This commit is contained in:
proxyctl coordinator
2026-07-17 22:23:31 +08:00
parent 177bf34354
commit 7c0b4481d7
17 changed files with 985 additions and 1013 deletions
+81 -66
View File
@@ -53,15 +53,6 @@ type TargetSnapshot struct {
Evidence string
}
// TargetDetectionUpdate carries the complete authoritative Application state
// after one Target changed. TargetID identifies that change without duplicating
// availability, inspection, or failure fields already present in Snapshot.
type TargetDetectionUpdate struct {
Snapshot Snapshot
TargetID string
Complete bool
}
type Snapshot struct {
Profiles []profile.Profile
ActiveProfile string
@@ -70,9 +61,6 @@ type Snapshot struct {
type TargetProgress struct {
TargetID string
// Applying is retained for synchronous DispatchProgress callers. New
// interactive callers use the structured Phase field.
Applying bool
Phase TargetProgressPhase
State target.State
Evidence string
@@ -80,15 +68,12 @@ type TargetProgress struct {
}
type Result struct {
Snapshot Snapshot
Notice string
// Confirmation remains for existing synchronous callers.
Confirmation string
Snapshot Snapshot
Notice string
ConfirmationRequest *Confirmation
Status OperationStatus
Progress []TargetProgress
FirstFailedTargetID string
Complete bool
Err error
}
@@ -97,20 +82,6 @@ type Module interface {
Dispatch(Intent) Result
}
// ProgressiveModule is implemented by Applications that can expose each
// coordinated Target update while an Active Profile transition is running.
// The final Result has Complete set and closes the stream.
type ProgressiveModule interface {
DispatchProgress(Intent) <-chan Result
}
// ProgressiveDetectionModule exposes a useful all-Checking snapshot
// synchronously, followed by one structured Target result at a time. Startup
// and explicit refresh use this same presentation-independent contract.
type ProgressiveDetectionModule interface {
BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate)
}
type Application struct {
store profile.Store
state profile.State
@@ -196,7 +167,13 @@ type targetDetectionOutcome struct {
detection target.Detection
}
func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate) {
type targetDetectionUpdate struct {
snapshot Snapshot
targetID string
complete bool
}
func (a *Application) beginTargetDetection() (Snapshot, <-chan targetDetectionUpdate) {
a.stateMu.Lock()
a.detectionGeneration++
generation := a.detectionGeneration
@@ -208,7 +185,7 @@ func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUp
checking := a.snapshotLocked()
a.stateMu.Unlock()
updates := make(chan TargetDetectionUpdate, len(a.targets)+1)
updates := make(chan targetDetectionUpdate, len(a.targets)+1)
outcomes := make(chan targetDetectionOutcome, len(a.targets))
for i, adapter := range a.targets {
adapter, id := adapter, a.targetIDs[i]
@@ -233,7 +210,7 @@ func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUp
a.inspections[outcome.targetID] = outcome.detection.Inspection
a.detectionPhase[outcome.targetID] = phase
a.detectionErr[outcome.targetID] = outcome.detection.Err
update := TargetDetectionUpdate{Snapshot: a.snapshotLocked(), TargetID: outcome.targetID}
update := targetDetectionUpdate{snapshot: a.snapshotLocked(), targetID: outcome.targetID}
// The buffered send cannot wait on a consumer and happens before a
// newer Begin call can make this generation stale.
updates <- update
@@ -241,7 +218,7 @@ func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUp
}
a.stateMu.Lock()
if generation == a.detectionGeneration {
updates <- TargetDetectionUpdate{Snapshot: a.snapshotLocked(), Complete: true}
updates <- targetDetectionUpdate{snapshot: a.snapshotLocked(), complete: true}
}
a.stateMu.Unlock()
}()
@@ -262,17 +239,6 @@ func (a *Application) Dispatch(intent Intent) Result {
return a.dispatch(intent, nil)
}
func (a *Application) DispatchProgress(intent Intent) <-chan Result {
results := make(chan Result, 2*len(a.targets)+1)
go func() {
defer close(results)
final := a.dispatch(intent, func(update Result) { results <- update })
final.Complete = true
results <- final
}()
return results
}
func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
// Serialize local mutation workflows even when tests intentionally omit the
// cross-process MutationLock.
@@ -302,8 +268,7 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
return a.setTargetDesired(intent, publish)
}
if intent.Kind == RefreshTargets {
a.refreshTargets()
return Result{Snapshot: a.Snapshot(), Notice: "Refreshed Target availability and state"}
return a.refreshTargets(publish)
}
if intent.Kind == ActivateProfile {
return a.activateProfile(intent, publish)
@@ -371,8 +336,9 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
Title: "Confirm Active Profile change", Summary: "Changing the Active Profile has high-impact Target effects.",
Details: warnings, ConfirmLabel: "Change Profile", CancelLabel: "Cancel", HighImpact: true,
}
return Result{Snapshot: a.Snapshot(), Confirmation: message, ConfirmationRequest: confirmation,
Status: OperationStatus{Kind: StatusConfirmation, Summary: confirmation.Title, Detail: confirmation.Summary}}
_ = message
return Result{Snapshot: a.Snapshot(), ConfirmationRequest: confirmation,
Status: OperationStatus{Summary: confirmation.Title}}
}
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
@@ -410,8 +376,8 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
name := a.targetDescriptions[targetIndex].Name
updated[id] = true
if publish != nil {
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: id, Applying: true, Phase: TargetProgressRunning, State: a.inspection(id).State}}, Notice: "Updating " + name + " Target",
Status: OperationStatus{Kind: StatusProgress, Summary: "Updating " + name + " Target", TargetID: id}})
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: id, Phase: TargetProgressRunning, State: a.inspection(id).State}}, Notice: "Updating " + name + " Target",
Status: OperationStatus{Summary: "Updating " + name + " Target", TargetID: id}})
}
operation := adapter.SetDesired(context.Background(), true, active)
inspection := operation.Inspection
@@ -430,9 +396,9 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
item := TargetProgress{TargetID: id, Phase: progressPhase, State: inspection.State, Evidence: inspection.Evidence, Err: safeErr}
progress = append(progress, item)
if publish != nil {
status := OperationStatus{Kind: StatusSuccess, Summary: name + " Target update completed", TargetID: id}
status := OperationStatus{Summary: name + " Target update completed", TargetID: id}
if safeErr != nil {
status = OperationStatus{Kind: StatusError, Summary: name + " Target update failed", Detail: safeErr.Error(), TargetID: id}
status = OperationStatus{Summary: name + " Target update failed", TargetID: id}
}
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{item}, Notice: name + " Target update completed", Status: status, Err: safeErr})
}
@@ -458,11 +424,11 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
}
a.mergeInspections(remaining)
result := Result{Snapshot: a.Snapshot(), Progress: progress, Notice: fmt.Sprintf("Activated Profile %q", intent.Name),
Status: OperationStatus{Kind: StatusSuccess, Summary: fmt.Sprintf("Activated Profile %q", intent.Name)}}
Status: OperationStatus{Summary: fmt.Sprintf("Activated Profile %q", intent.Name)}}
if len(failures) > 0 {
result.Err = errors.Join(failures...)
result.Notice = fmt.Sprintf("Activated Profile %q with %d Target update failure(s); successful updates were kept and no global rollback was attempted", intent.Name, len(failures))
result.Status = OperationStatus{Kind: StatusError, Summary: fmt.Sprintf("Activated Profile %q with %d Target update failure(s)", intent.Name, len(failures)), Detail: result.Err.Error()}
result.Status = OperationStatus{Summary: fmt.Sprintf("Activated Profile %q with %d Target update failure(s)", intent.Name, len(failures))}
for _, item := range progress {
if item.Err != nil {
result.FirstFailedTargetID = item.TargetID
@@ -548,10 +514,61 @@ func (a *Application) mergeInspections(inspections map[string]target.Inspection)
}
}
func (a *Application) refreshTargets() {
_, updates := a.BeginTargetDetection()
for range updates {
func targetByID(snapshot Snapshot, id string) (TargetSnapshot, string) {
for _, item := range snapshot.Targets {
if item.ID == id {
name := item.Description.Name
if name == "" {
name = id
}
return item, name
}
}
return TargetSnapshot{ID: id}, id
}
func (a *Application) refreshTargets(publish func(Result)) Result {
checking, updates := a.beginTargetDetection()
if publish != nil {
progress := make([]TargetProgress, 0, len(checking.Targets))
for _, item := range checking.Targets {
progress = append(progress, TargetProgress{TargetID: item.ID, Phase: TargetProgressRunning, State: item.State})
}
publish(Result{Snapshot: checking, Status: OperationStatus{Summary: "Checking Target availability and state"}, Progress: progress})
}
var failures []error
firstFailed := ""
final := checking
for update := range updates {
final = update.snapshot
if update.complete {
continue
}
item, name := targetByID(update.snapshot, update.targetID)
progress := TargetProgress{TargetID: update.targetID, Phase: TargetProgressSucceeded, State: item.State, Evidence: item.Evidence}
status := OperationStatus{Summary: "Checked " + name + " Target", TargetID: update.targetID}
if item.DetectionPhase == TargetDetectionFailed {
progress.Phase = TargetProgressFailed
progress.Err = item.DetectionErr
status.Summary = "Could not check " + name + " Target"
if item.DetectionErr != nil {
failures = append(failures, fmt.Errorf("%s Target: %w", name, item.DetectionErr))
}
if firstFailed == "" {
firstFailed = update.targetID
}
}
if publish != nil {
publish(Result{Snapshot: update.snapshot, Status: status, Progress: []TargetProgress{progress}, Err: progress.Err})
}
}
result := Result{Snapshot: final, Notice: "Target availability and state checks completed", Status: OperationStatus{Summary: "Target availability and state checks completed"}}
if len(failures) > 0 {
result.Err = errors.Join(failures...)
result.FirstFailedTargetID = firstFailed
result.Status = OperationStatus{Summary: "Some Target checks failed", TargetID: firstFailed}
}
return result
}
func (a *Application) refreshInspections() {
@@ -582,12 +599,12 @@ func (a *Application) setTargetDesired(intent Intent, publish func(Result)) Resu
if warning := selected.Confirmation(intent.Enabled); warning != "" && !intent.Confirm {
confirmation := &Confirmation{Title: "Confirm " + a.targetDescriptions[selectedIndex].Name + " Target change", Summary: warning,
ConfirmLabel: "Continue", CancelLabel: "Cancel", HighImpact: true}
return Result{Snapshot: a.Snapshot(), Confirmation: warning, ConfirmationRequest: confirmation,
Status: OperationStatus{Kind: StatusConfirmation, Summary: confirmation.Title, Detail: confirmation.Summary, TargetID: intent.TargetID}}
return Result{Snapshot: a.Snapshot(), ConfirmationRequest: confirmation,
Status: OperationStatus{Summary: confirmation.Title, TargetID: intent.TargetID}}
}
if publish != nil {
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: intent.TargetID, Applying: true, Phase: TargetProgressRunning, State: a.inspection(intent.TargetID).State}},
Status: OperationStatus{Kind: StatusProgress, Summary: "Updating " + a.targetDescriptions[selectedIndex].Name + " Target", TargetID: intent.TargetID}})
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: intent.TargetID, Phase: TargetProgressRunning, State: a.inspection(intent.TargetID).State}},
Status: OperationStatus{Summary: "Updating " + a.targetDescriptions[selectedIndex].Name + " Target", TargetID: intent.TargetID}})
}
result := selected.SetDesired(context.Background(), intent.Enabled, active)
inspection := result.Inspection
@@ -608,15 +625,13 @@ func (a *Application) setTargetDesired(intent Intent, publish func(Result)) Resu
}
notice := ""
progressPhase := TargetProgressSucceeded
status := OperationStatus{Kind: StatusSuccess, TargetID: intent.TargetID}
status := OperationStatus{TargetID: intent.TargetID}
if result.Err == nil {
notice = fmt.Sprintf("%s Target is now %s", a.targetDescriptions[selectedIndex].Name, inspection.State)
status.Summary = notice
} else {
progressPhase = TargetProgressFailed
status.Kind = StatusError
status.Summary = a.targetDescriptions[selectedIndex].Name + " Target update failed"
status.Detail = result.Err.Error()
}
progress := TargetProgress{TargetID: intent.TargetID, Phase: progressPhase, State: inspection.State, Evidence: inspection.Evidence, Err: result.Err}
return Result{Snapshot: a.Snapshot(), Notice: notice, Status: status, Progress: []TargetProgress{progress}, FirstFailedTargetID: map[bool]string{true: intent.TargetID}[result.Err != nil], Err: result.Err}
+26 -20
View File
@@ -126,27 +126,31 @@ func TestActiveProfileSwitchCoordinatesEnabledTargetsWithoutRollback(t *testing.
}
inspectCallsBeforeSwitch := second.inspectCalls
var result app.Result
var failedCompletion app.Result
for update := range application.DispatchProgress(app.Intent{Kind: app.ActivateProfile, Name: "new"}) {
result = update
if len(update.Progress) == 1 && update.Progress[0].TargetID == "second" && !update.Progress[0].Applying {
var final app.OperationUpdate
var failedCompletion app.OperationUpdate
stream := application.StartOperation(app.Intent{Kind: app.ActivateProfile, Name: "new"})
for update := range stream.Updates {
final = update
progress := update.Progress()
if len(progress) == 1 && progress[0].TargetID == "second" && progress[0].Phase == app.TargetProgressFailed {
failedCompletion = update
}
}
if result.Snapshot.ActiveProfile != "new" {
t.Fatalf("Active Profile = %q", result.Snapshot.ActiveProfile)
if final.Snapshot.ActiveProfile != "new" {
t.Fatalf("Active Profile = %q", final.Snapshot.ActiveProfile)
}
if result.Err == nil || strings.Contains(result.Err.Error(), "secret") || !strings.Contains(result.Err.Error(), "second") {
t.Fatalf("redacted partial failure = %v", result.Err)
if final.Err() == nil || strings.Contains(final.Err().Error(), "secret") || !strings.Contains(final.Err().Error(), "second") {
t.Fatalf("redacted partial failure = %v", final.Err())
}
if got := result.Snapshot.Targets; got[0].State != target.Enabled || got[1].State != target.Drifted {
if got := final.Snapshot.Targets; got[0].State != target.Enabled || got[1].State != target.Drifted {
t.Fatalf("truthful partial states = %#v", got)
}
if len(result.Progress) != 2 || result.Progress[0].TargetID != "first" || result.Progress[0].State != target.Enabled || result.Progress[1].TargetID != "second" || result.Progress[1].State != target.Drifted || result.Progress[1].Evidence != "fresh actual configuration" {
t.Fatalf("per-Target progress = %#v", result.Progress)
progress := final.Progress()
if len(progress) != 2 || progress[0].TargetID != "first" || progress[0].State != target.Enabled || progress[1].TargetID != "second" || progress[1].State != target.Drifted || progress[1].Evidence != "fresh actual configuration" {
t.Fatalf("per-Target progress = %#v", progress)
}
if len(failedCompletion.Progress) != 1 || failedCompletion.Progress[0].State != target.Drifted || failedCompletion.Progress[0].Evidence != "fresh actual configuration" || failedCompletion.Err == nil {
failedProgress := failedCompletion.Progress()
if len(failedProgress) != 1 || failedProgress[0].State != target.Drifted || failedProgress[0].Evidence != "fresh actual configuration" || failedProgress[0].Err == nil {
t.Fatalf("failed completion event was not truthful: %#v", failedCompletion)
}
if failedCompletion.Snapshot.Targets[1].State != target.Drifted || failedCompletion.Snapshot.Targets[1].Evidence != "fresh actual configuration" {
@@ -155,8 +159,8 @@ func TestActiveProfileSwitchCoordinatesEnabledTargetsWithoutRollback(t *testing.
if second.inspectCalls < inspectCallsBeforeSwitch+2 {
t.Fatalf("failed SetDesired was not explicitly re-inspected: Inspect calls before=%d after=%d", inspectCallsBeforeSwitch, second.inspectCalls)
}
if !strings.Contains(result.Err.Error(), "fresh actual configuration") || !strings.Contains(result.Err.Error(), "Drifted") {
t.Fatalf("failure summary omitted fresh inspection: %v", result.Err)
if !strings.Contains(final.Err().Error(), "fresh actual configuration") || !strings.Contains(final.Err().Error(), "Drifted") {
t.Fatalf("failure summary omitted fresh inspection: %v", final.Err())
}
if len(first.applied) != 2 || first.applied[1] != "new" {
t.Fatalf("successful Target was not updated: %#v", first.applied)
@@ -231,7 +235,7 @@ func TestActiveProfileProgressReusesAvailabilityCache(t *testing.T) {
callsBeforeSwitch := adapter.availabilityCalls
var updates int
for range application.DispatchProgress(app.Intent{Kind: app.ActivateProfile, Name: "new"}) {
for range application.StartOperation(app.Intent{Kind: app.ActivateProfile, Name: "new"}).Updates {
updates++
}
if updates < 3 {
@@ -264,17 +268,19 @@ func TestActiveProfileSwitchStreamsApplyingAndCompletionForEachTarget(t *testing
t.Fatal(result.Err)
}
updates := application.DispatchProgress(app.Intent{Kind: app.ActivateProfile, Name: "new"})
updates := application.StartOperation(app.Intent{Kind: app.ActivateProfile, Name: "new"}).Updates
applying := <-updates
completed := <-updates
final := <-updates
if len(applying.Progress) != 1 || !applying.Progress[0].Applying || applying.Progress[0].TargetID != "shell" {
applyingProgress := applying.Progress()
if len(applyingProgress) != 1 || applyingProgress[0].Phase != app.TargetProgressRunning || applyingProgress[0].TargetID != "shell" {
t.Fatalf("applying update = %#v", applying)
}
if len(completed.Progress) != 1 || completed.Progress[0].Applying || completed.Progress[0].State != target.Enabled {
completedProgress := completed.Progress()
if len(completedProgress) != 1 || completedProgress[0].Phase != app.TargetProgressSucceeded || completedProgress[0].State != target.Enabled {
t.Fatalf("completion update = %#v", completed)
}
if !final.Complete || final.Snapshot.ActiveProfile != "new" {
if !final.IsSucceeded() || final.Snapshot.ActiveProfile != "new" {
t.Fatalf("final update = %#v", final)
}
}
+176 -138
View File
@@ -1,21 +1,5 @@
package app
import (
"errors"
"fmt"
)
// OperationPhase describes coordination state independently of any terminal UI.
type OperationPhase uint8
const (
OperationRunning OperationPhase = iota + 1
OperationAwaitingConfirmation
OperationSucceeded
OperationFailed
OperationBlocked
)
// StatusKind gives presentation layers semantics without requiring them to
// inspect user-facing status text.
type StatusKind uint8
@@ -25,15 +9,12 @@ const (
StatusSuccess
StatusError
StatusBlocked
StatusConfirmation
)
// OperationStatus is the current concise status plus optional expandable
// detail. TargetID relates a result to the Target it can explicitly resolve.
// OperationStatus is concise status text related to an optional Target.
type OperationStatus struct {
Kind StatusKind
Summary string
Detail string
TargetID string
}
@@ -57,21 +38,161 @@ const (
TargetProgressFailed
)
// OperationUpdate is the common progressive contract used by interactive
// detection, Target mutation, and Active Profile transition operations.
type OperationUpdate struct {
Snapshot Snapshot
Phase OperationPhase
Status OperationStatus
Progress []TargetProgress
Confirmation *Confirmation
FirstFailedTargetID string
Err error
// OperationState is a closed tagged set. Its concrete value makes an update
// exactly one of running, confirmation, succeeded, failed, or closed.
type OperationState interface {
operationState()
}
type operationRunning struct {
status OperationStatus
progress []TargetProgress
}
func (operationRunning) operationState() {}
type operationConfirmation struct{ request Confirmation }
func (operationConfirmation) operationState() {}
type operationSucceeded struct {
status OperationStatus
progress []TargetProgress
}
func (operationSucceeded) operationState() {}
type operationFailed struct {
status OperationStatus
progress []TargetProgress
firstFailedTargetID string
err error
}
func (operationFailed) operationState() {}
type operationClosed struct{ status OperationStatus }
func (operationClosed) operationState() {}
// OperationUpdate is the common progressive contract used by interactive
// detection, Target mutation, and Active Profile transition operations. Build
// values with the constructors below so contradictory states are unrepresentable.
type OperationUpdate struct {
Snapshot Snapshot
State OperationState
}
func RunningOperationUpdate(snapshot Snapshot, status OperationStatus, progress []TargetProgress) OperationUpdate {
status.Kind = StatusProgress
return OperationUpdate{Snapshot: snapshot, State: operationRunning{status: status, progress: cloneProgress(progress)}}
}
func ConfirmationOperationUpdate(snapshot Snapshot, request Confirmation) OperationUpdate {
return OperationUpdate{Snapshot: snapshot, State: operationConfirmation{request: request}}
}
func SucceededOperationUpdate(snapshot Snapshot, status OperationStatus, progress []TargetProgress) OperationUpdate {
status.Kind = StatusSuccess
return OperationUpdate{Snapshot: snapshot, State: operationSucceeded{status: status, progress: cloneProgress(progress)}}
}
func FailedOperationUpdate(snapshot Snapshot, status OperationStatus, progress []TargetProgress, firstFailedTargetID string, err error) OperationUpdate {
status.Kind = StatusError
return OperationUpdate{Snapshot: snapshot, State: operationFailed{
status: status, progress: cloneProgress(progress), firstFailedTargetID: firstFailedTargetID, err: err,
}}
}
// ClosedOperationUpdate reports a request that cannot start, such as an
// overlapping interactive operation. It is terminal and has no progress or
// error payload.
func ClosedOperationUpdate(snapshot Snapshot, status OperationStatus) OperationUpdate {
status.Kind = StatusBlocked
return OperationUpdate{Snapshot: snapshot, State: operationClosed{status: status}}
}
func cloneProgress(progress []TargetProgress) []TargetProgress {
return append([]TargetProgress(nil), progress...)
}
func (u OperationUpdate) Status() OperationStatus {
switch state := u.State.(type) {
case operationRunning:
return state.status
case operationSucceeded:
return state.status
case operationFailed:
return state.status
case operationClosed:
return state.status
default:
return OperationStatus{}
}
}
func (u OperationUpdate) Progress() []TargetProgress {
switch state := u.State.(type) {
case operationRunning:
return cloneProgress(state.progress)
case operationSucceeded:
return cloneProgress(state.progress)
case operationFailed:
return cloneProgress(state.progress)
default:
return nil
}
}
func (u OperationUpdate) Confirmation() *Confirmation {
state, ok := u.State.(operationConfirmation)
if !ok {
return nil
}
request := state.request
request.Details = append([]string(nil), request.Details...)
return &request
}
func (u OperationUpdate) Err() error {
if state, ok := u.State.(operationFailed); ok {
return state.err
}
return nil
}
func (u OperationUpdate) FirstFailedTargetID() string {
if state, ok := u.State.(operationFailed); ok {
return state.firstFailedTargetID
}
return ""
}
func (u OperationUpdate) IsRunning() bool {
_, ok := u.State.(operationRunning)
return ok
}
func (u OperationUpdate) IsConfirmation() bool {
_, ok := u.State.(operationConfirmation)
return ok
}
func (u OperationUpdate) IsSucceeded() bool {
_, ok := u.State.(operationSucceeded)
return ok
}
func (u OperationUpdate) IsFailed() bool {
_, ok := u.State.(operationFailed)
return ok
}
func (u OperationUpdate) IsTerminal() bool { return !u.IsRunning() }
// OperationStream exposes the useful initial frame synchronously and all
// subsequent structured changes through Updates. Updates closes after a
// terminal phase.
// terminal update.
type OperationStream struct {
Snapshot Snapshot
Updates <-chan OperationUpdate
@@ -84,84 +205,32 @@ type InteractiveModule interface {
StartOperation(Intent) OperationStream
}
// StartOperation rejects overlap instead of queuing hidden interactive work.
// Synchronous Dispatch callers retain the established serialized behavior.
// StartOperation returns before dispatch serialization is acquired. This keeps
// Bubble Tea Update non-blocking even when a synchronous caller currently owns
// the serialized Application entry.
func (a *Application) StartOperation(intent Intent) OperationStream {
initial := a.Snapshot()
a.interactiveMu.Lock()
if a.interactiveOperation {
a.interactiveMu.Unlock()
updates := make(chan OperationUpdate, 1)
updates <- OperationUpdate{
Snapshot: a.Snapshot(),
Phase: OperationBlocked,
Status: OperationStatus{Kind: StatusBlocked,
Summary: "Another system operation is already in progress",
Detail: "Wait for the current operation to finish before starting another modification."},
}
updates <- ClosedOperationUpdate(initial, OperationStatus{
Summary: "Another system operation is already in progress",
})
close(updates)
return OperationStream{Snapshot: a.Snapshot(), Updates: updates}
return OperationStream{Snapshot: initial, Updates: updates}
}
a.interactiveOperation = true
a.interactiveMu.Unlock()
if intent.Kind == RefreshTargets {
return a.startDetectionOperation()
}
initial := a.Snapshot()
updates := make(chan OperationUpdate, 2*len(a.targets)+2)
go func() {
defer close(updates)
defer a.finishInteractiveOperation()
final := a.dispatch(intent, func(result Result) {
updates <- operationUpdate(result, false)
updates <- runningOperationUpdate(result)
})
updates <- operationUpdate(final, true)
}()
return OperationStream{Snapshot: initial, Updates: updates}
}
func (a *Application) startDetectionOperation() OperationStream {
initial, detection := a.BeginTargetDetection()
updates := make(chan OperationUpdate, len(a.targets)+1)
go func() {
defer close(updates)
defer a.finishInteractiveOperation()
var failures []error
firstFailed := ""
for item := range detection {
if item.Complete {
update := OperationUpdate{
Snapshot: item.Snapshot,
Phase: OperationSucceeded,
Status: OperationStatus{Kind: StatusSuccess, Summary: "Target availability and state checks completed"},
}
if len(failures) > 0 {
update.Phase = OperationFailed
update.Err = errors.Join(failures...)
update.FirstFailedTargetID = firstFailed
update.Status = OperationStatus{Kind: StatusError, Summary: "Some Target checks failed", Detail: update.Err.Error(), TargetID: firstFailed}
}
updates <- update
continue
}
targetSnapshot, name := targetByID(item.Snapshot, item.TargetID)
progress := TargetProgress{TargetID: item.TargetID, State: targetSnapshot.State, Evidence: targetSnapshot.Evidence, Phase: TargetProgressSucceeded}
status := OperationStatus{Kind: StatusProgress, Summary: "Checked " + name + " Target", TargetID: item.TargetID}
if targetSnapshot.DetectionPhase == TargetDetectionFailed {
progress.Phase = TargetProgressFailed
progress.Err = targetSnapshot.DetectionErr
status = OperationStatus{Kind: StatusError, Summary: "Could not check " + name + " Target", TargetID: item.TargetID}
if targetSnapshot.DetectionErr != nil {
status.Detail = targetSnapshot.DetectionErr.Error()
failures = append(failures, fmt.Errorf("%s Target: %w", name, targetSnapshot.DetectionErr))
}
if firstFailed == "" {
firstFailed = item.TargetID
}
}
updates <- OperationUpdate{Snapshot: item.Snapshot, Phase: OperationRunning, Status: status, Progress: []TargetProgress{progress}, Err: progress.Err}
}
updates <- terminalOperationUpdate(final)
}()
return OperationStream{Snapshot: initial, Updates: updates}
}
@@ -172,57 +241,26 @@ func (a *Application) finishInteractiveOperation() {
a.interactiveMu.Unlock()
}
func targetByID(snapshot Snapshot, id string) (TargetSnapshot, string) {
for _, item := range snapshot.Targets {
if item.ID == id {
name := item.Description.Name
if name == "" {
name = id
}
return item, name
}
func runningOperationUpdate(result Result) OperationUpdate {
if result.ConfirmationRequest != nil {
return ConfirmationOperationUpdate(result.Snapshot, *result.ConfirmationRequest)
}
return TargetSnapshot{ID: id}, id
return RunningOperationUpdate(result.Snapshot, result.Status, result.Progress)
}
func operationUpdate(result Result, terminal bool) OperationUpdate {
update := OperationUpdate{
Snapshot: result.Snapshot,
Phase: OperationRunning,
Status: result.Status,
Progress: result.Progress,
Confirmation: result.ConfirmationRequest,
FirstFailedTargetID: result.FirstFailedTargetID,
Err: result.Err,
}
func terminalOperationUpdate(result Result) OperationUpdate {
if result.ConfirmationRequest != nil {
update.Phase = OperationAwaitingConfirmation
if update.Status.Kind == 0 {
update.Status = OperationStatus{Kind: StatusConfirmation, Summary: result.ConfirmationRequest.Title, Detail: result.ConfirmationRequest.Summary}
}
return update
}
if update.Status.Kind == 0 {
switch {
case result.Err != nil:
update.Status = OperationStatus{Kind: StatusError, Summary: "Operation failed", Detail: result.Err.Error()}
case result.Notice != "":
update.Status = OperationStatus{Kind: StatusSuccess, Summary: result.Notice}
}
}
if !terminal {
return update
return ConfirmationOperationUpdate(result.Snapshot, *result.ConfirmationRequest)
}
status := result.Status
if result.Err != nil {
update.Phase = OperationFailed
if update.Status.Kind != StatusError {
update.Status = OperationStatus{Kind: StatusError, Summary: "Operation failed", Detail: result.Err.Error(), TargetID: update.Status.TargetID}
}
} else {
update.Phase = OperationSucceeded
if update.Status.Kind == StatusProgress || update.Status.Kind == 0 {
update.Status.Kind = StatusSuccess
if status.Summary == "" {
status.Summary = "Operation failed"
}
return FailedOperationUpdate(result.Snapshot, status, result.Progress, result.FirstFailedTargetID, result.Err)
}
return update
if status.Summary == "" {
status.Summary = result.Notice
}
return SucceededOperationUpdate(result.Snapshot, status, result.Progress)
}
+122 -11
View File
@@ -5,7 +5,9 @@ import (
"errors"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"proxyctl/internal/app"
"proxyctl/internal/profile"
@@ -13,12 +15,17 @@ import (
)
type coordinatedOperationTarget struct {
id string
state target.State
started chan struct{}
release chan struct{}
confirmation string
failure error
id string
state target.State
started chan struct{}
release chan struct{}
detectionStarted chan struct{}
detectionRelease chan struct{}
confirmation string
failure error
mu sync.Mutex
inDetection bool
overlapObserved chan bool
}
func (t *coordinatedOperationTarget) ID() string { return t.id }
@@ -32,7 +39,28 @@ func (t *coordinatedOperationTarget) Inspect(context.Context, *profile.Profile)
return target.Inspection{State: t.state, Evidence: "fresh state"}
}
func (t *coordinatedOperationTarget) Confirmation(bool) string { return t.confirmation }
func (t *coordinatedOperationTarget) Detect(ctx context.Context, active *profile.Profile) target.Detection {
t.mu.Lock()
t.inDetection = true
t.mu.Unlock()
if t.detectionStarted != nil {
close(t.detectionStarted)
}
if t.detectionRelease != nil {
<-t.detectionRelease
}
t.mu.Lock()
t.inDetection = false
t.mu.Unlock()
return target.Detection{Availability: t.Availability(ctx), Inspection: t.Inspect(ctx, active)}
}
func (t *coordinatedOperationTarget) SetDesired(_ context.Context, enabled bool, _ *profile.Profile) target.OperationResult {
t.mu.Lock()
overlapped := t.inDetection
t.mu.Unlock()
if t.overlapObserved != nil {
t.overlapObserved <- overlapped
}
if t.started != nil {
close(t.started)
}
@@ -66,14 +94,15 @@ func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgr
operation := application.StartOperation(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
running := <-operation.Updates
if running.Phase != app.OperationRunning || len(running.Progress) != 1 || running.Progress[0].Phase != app.TargetProgressRunning || running.Progress[0].TargetID != "shell" || running.Status.Kind != app.StatusProgress {
runningProgress := running.Progress()
if !running.IsRunning() || len(runningProgress) != 1 || runningProgress[0].Phase != app.TargetProgressRunning || runningProgress[0].TargetID != "shell" || running.Status().Kind != app.StatusProgress {
t.Fatalf("running update = %#v", running)
}
<-adapter.started
blocked := application.StartOperation(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: false})
blockedUpdate := <-blocked.Updates
if blockedUpdate.Phase != app.OperationBlocked || blockedUpdate.Status.Kind != app.StatusBlocked || !strings.Contains(blockedUpdate.Status.Summary, "already in progress") {
if !blockedUpdate.IsTerminal() || blockedUpdate.Status().Kind != app.StatusBlocked || !strings.Contains(blockedUpdate.Status().Summary, "already in progress") {
t.Fatalf("blocked update = %#v", blockedUpdate)
}
if _, open := <-blocked.Updates; open {
@@ -82,7 +111,8 @@ func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgr
close(adapter.release)
final := <-operation.Updates
if final.Phase != app.OperationSucceeded || final.Snapshot.Targets[0].State != target.Enabled || len(final.Progress) != 1 || final.Progress[0].Phase != app.TargetProgressSucceeded {
finalProgress := final.Progress()
if !final.IsSucceeded() || final.Snapshot.Targets[0].State != target.Enabled || len(finalProgress) != 1 || finalProgress[0].Phase != app.TargetProgressSucceeded {
t.Fatalf("final update = %#v", final)
}
if _, open := <-operation.Updates; open {
@@ -90,6 +120,86 @@ func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgr
}
}
func TestDetectionOperationSerializesWithSynchronousMutation(t *testing.T) {
adapter := &coordinatedOperationTarget{
id: "shell", state: target.Disabled,
detectionStarted: make(chan struct{}), detectionRelease: make(chan struct{}),
started: make(chan struct{}), overlapObserved: make(chan bool, 1),
}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
}
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
t.Fatal(result.Err)
}
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); result.Err != nil {
t.Fatal(result.Err)
}
detection := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
<-adapter.detectionStarted
mutationAttempted := make(chan struct{})
mutationDone := make(chan app.Result, 1)
go func() {
close(mutationAttempted)
mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
}()
<-mutationAttempted
close(adapter.detectionRelease)
for range detection.Updates {
}
result := <-mutationDone
if result.Err != nil || result.Snapshot.Targets[0].State != target.Enabled {
t.Fatalf("serialized mutation = %#v", result)
}
if overlapped := <-adapter.overlapObserved; overlapped {
t.Fatal("mutation overlapped Target detection")
}
}
func TestStartOperationReturnsBeforeDispatchSerializationIsAvailable(t *testing.T) {
adapter := &coordinatedOperationTarget{
id: "shell", state: target.Disabled,
started: make(chan struct{}), release: make(chan struct{}),
detectionStarted: make(chan struct{}), detectionRelease: make(chan struct{}),
}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
}
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
t.Fatal(result.Err)
}
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); result.Err != nil {
t.Fatal(result.Err)
}
mutationDone := make(chan app.Result, 1)
go func() {
mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
}()
<-adapter.started // barrier: synchronous Dispatch owns dispatch serialization
returned := make(chan app.OperationStream, 1)
go func() { returned <- application.StartOperation(app.Intent{Kind: app.RefreshTargets}) }()
var stream app.OperationStream
select {
case stream = <-returned:
case <-time.After(time.Second):
t.Fatal("StartOperation blocked Bubble Tea while dispatch serialization was held")
}
close(adapter.release)
if result := <-mutationDone; result.Err != nil {
t.Fatal(result.Err)
}
<-adapter.detectionStarted
close(adapter.detectionRelease)
for range stream.Updates {
}
}
func TestInteractiveOperationStreamUsesStructuredConfirmation(t *testing.T) {
adapter := &coordinatedOperationTarget{id: "docker-daemon", state: target.Disabled, confirmation: "Docker will restart and may disrupt running work."}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
@@ -105,7 +215,8 @@ func TestInteractiveOperationStreamUsesStructuredConfirmation(t *testing.T) {
operation := application.StartOperation(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
update := <-operation.Updates
if update.Phase != app.OperationAwaitingConfirmation || update.Confirmation == nil || update.Confirmation.Title == "" || update.Confirmation.CancelLabel == "" || update.Status.Kind != app.StatusConfirmation {
confirmation := update.Confirmation()
if !update.IsConfirmation() || confirmation == nil || confirmation.Title == "" || confirmation.CancelLabel == "" {
t.Fatalf("confirmation update = %#v", update)
}
if _, open := <-operation.Updates; open {
@@ -140,7 +251,7 @@ func TestInteractiveProfileTransitionReportsFirstFailureWithoutRollback(t *testi
for update := range operation.Updates {
final = update
}
if final.Phase != app.OperationFailed || final.FirstFailedTargetID != "second" || final.Snapshot.ActiveProfile != "new" {
if !final.IsFailed() || final.FirstFailedTargetID() != "second" || final.Snapshot.ActiveProfile != "new" {
t.Fatalf("final transition = %#v", final)
}
if final.Snapshot.Targets[0].State != target.Enabled || final.Snapshot.Targets[1].State != target.Drifted {
+40 -37
View File
@@ -60,16 +60,22 @@ func TestApplicationStreamsIndependentTargetDetectionInProductOrder(t *testing.T
}
}
checking, updates := application.BeginTargetDetection()
stream := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
checking := stream.Snapshot
for _, item := range checking.Targets {
if item.DetectionPhase != app.TargetChecking {
t.Fatalf("Target did not begin in Checking: %#v", item)
}
}
updates := stream.Updates
initial := <-updates
if !initial.IsRunning() || len(initial.Progress()) != 3 {
t.Fatalf("initial detection progress = %#v", initial)
}
close(secondRelease)
second := <-updates
if second.TargetID != "second" || second.Snapshot.Targets[1].DetectionPhase != app.TargetReady || second.Snapshot.Targets[1].State != target.External {
if second.Status().TargetID != "second" || second.Snapshot.Targets[1].DetectionPhase != app.TargetReady || second.Snapshot.Targets[1].State != target.External {
t.Fatalf("second completion = %#v", second)
}
assertOnlyDetectionChanged(t, checking, second.Snapshot, "second")
@@ -77,20 +83,20 @@ func TestApplicationStreamsIndependentTargetDetectionInProductOrder(t *testing.T
close(failedRelease)
failed := <-updates
if failed.TargetID != "failed" || failed.Snapshot.Targets[2].DetectionPhase != app.TargetDetectionFailed || failed.Snapshot.Targets[2].DetectionErr == nil {
if failed.Status().TargetID != "failed" || failed.Snapshot.Targets[2].DetectionPhase != app.TargetDetectionFailed || failed.Snapshot.Targets[2].DetectionErr == nil {
t.Fatalf("failed completion = %#v", failed)
}
assertOnlyDetectionChanged(t, second.Snapshot, failed.Snapshot, "failed")
close(firstRelease)
first := <-updates
if first.TargetID != "first" || first.Snapshot.Targets[0].DetectionPhase != app.TargetReady || first.Snapshot.Targets[0].State != target.Enabled {
if first.Status().TargetID != "first" || first.Snapshot.Targets[0].DetectionPhase != app.TargetReady || first.Snapshot.Targets[0].State != target.Enabled {
t.Fatalf("first completion = %#v", first)
}
assertOnlyDetectionChanged(t, failed.Snapshot, first.Snapshot, "first")
final := <-updates
if !final.Complete {
if !final.IsFailed() {
t.Fatalf("final update = %#v", final)
}
assertTargetOrder(t, final.Snapshot, []string{"first", "second", "failed"})
@@ -127,45 +133,36 @@ func (t *overlappingDetectionTarget) Detect(context.Context, *profile.Profile) t
return <-request.result
}
func TestOverlappingTargetDetectionDiscardsLateOlderGeneration(t *testing.T) {
adapter := &overlappingDetectionTarget{requests: make(chan detectionRequest, 2)}
func TestOverlappingTargetDetectionIsRejectedAtTheSerializedOperationEntry(t *testing.T) {
adapter := &overlappingDetectionTarget{requests: make(chan detectionRequest, 1)}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
}
_, olderUpdates := application.BeginTargetDetection()
older := <-adapter.requests
_, newerUpdates := application.BeginTargetDetection()
newer := <-adapter.requests
newer.result <- target.Detection{
Availability: target.Availability{Available: true, Evidence: "new availability"},
Inspection: target.Inspection{State: target.Enabled, Evidence: "new inspection"},
older := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
request := <-adapter.requests
newer := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
closed := <-newer.Updates
if closed.Status().Kind != app.StatusBlocked {
t.Fatalf("overlapping detection = %#v", closed)
}
for range newerUpdates {
request.result <- target.Detection{
Availability: target.Availability{Available: true, Evidence: "authoritative availability"},
Inspection: target.Inspection{State: target.Enabled, Evidence: "authoritative inspection"},
}
older.result <- target.Detection{
Availability: target.Availability{Evidence: "old availability"},
Inspection: target.Inspection{State: target.Disabled, Evidence: "old inspection"},
for range older.Updates {
}
staleUpdates := 0
for range olderUpdates {
staleUpdates++
}
if staleUpdates != 0 {
t.Fatalf("older generation published %d stale authoritative update(s)", staleUpdates)
}
got := application.Snapshot().Targets[0]
if !got.Available || got.State != target.Enabled || got.Evidence != "new inspection" {
t.Fatalf("late older generation overwrote newer result: %#v", got)
if !got.Available || got.State != target.Enabled {
t.Fatalf("serialized detection result = %#v", got)
}
}
type mutationInteractionTarget struct {
mu sync.Mutex
enabled bool
started chan struct{}
release chan struct{}
}
@@ -193,6 +190,7 @@ func (t *mutationInteractionTarget) SetDesired(context.Context, bool, *profile.P
return target.OperationResult{Inspection: target.Inspection{State: target.Enabled, Evidence: "mutation completed"}}
}
func (t *mutationInteractionTarget) Detect(context.Context, *profile.Profile) target.Detection {
close(t.started)
<-t.release
return target.Detection{
Availability: target.Availability{Available: true, Evidence: "stale detection"},
@@ -200,8 +198,8 @@ func (t *mutationInteractionTarget) Detect(context.Context, *profile.Profile) ta
}
}
func TestDispatchMutationInvalidatesInFlightTargetDetection(t *testing.T) {
adapter := &mutationInteractionTarget{release: make(chan struct{})}
func TestDispatchMutationWaitsForInFlightTargetDetection(t *testing.T) {
adapter := &mutationInteractionTarget{started: make(chan struct{}), release: make(chan struct{})}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
@@ -214,16 +212,21 @@ func TestDispatchMutationInvalidatesInFlightTargetDetection(t *testing.T) {
t.Fatal(activated.Err)
}
_, updates := application.BeginTargetDetection()
result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
detection := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
<-adapter.started
mutationDone := make(chan app.Result, 1)
go func() {
mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
}()
close(adapter.release)
for range detection.Updates {
}
result := <-mutationDone
if result.Err != nil || result.Snapshot.Targets[0].State != target.Enabled {
t.Fatalf("mutation result = %#v", result)
}
close(adapter.release)
for range updates {
}
if got := application.Snapshot().Targets[0]; got.State != target.Enabled || got.Evidence != "mutation completed" {
t.Fatalf("in-flight detection overwrote mutation: %#v", got)
t.Fatalf("serialized mutation state: %#v", got)
}
}
+1 -1
View File
@@ -85,7 +85,7 @@ func TestApplicationRequiresTargetConfirmationBeforeMutation(t *testing.T) {
createAndActivate(t, application)
warning := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "docker-daemon", Enabled: true})
if warning.Confirmation == "" || warning.Err != nil || len(adapter.calls) != 0 || warning.Snapshot.Targets[0].State != target.Disabled {
if warning.ConfirmationRequest == nil || warning.Err != nil || len(adapter.calls) != 0 || warning.Snapshot.Targets[0].State != target.Disabled {
t.Fatalf("unconfirmed result = %#v, calls=%v", warning, adapter.calls)
}
confirmed := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "docker-daemon", Enabled: true, Confirm: true})