gitea-admin/proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden operations and configuration recovery

Browse Source
This commit is contained in:
proxyctl coordinator
2026-08-30 11:13:45 +08:00
parent e34bf7d989
commit 8928986830
27 changed files with 542 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/app.go
+76 -51
View File
@@ -2,9 +2,10 @@ package app
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"strings"
"sync"
"time"
@@ -28,11 +29,21 @@ const (
)
type Intent struct {
Kind IntentKind
Name string
Profile profile.Profile
Confirm bool
TargetID string
Kind IntentKind
Name string
Profile profile.Profile
Confirm bool
ConfirmationToken string
TargetID string
}
func confirmationToken(parts ...string) string {
hash := sha256.New()
for _, part := range parts {
_, _ = hash.Write([]byte(part))
_, _ = hash.Write([]byte{0})
}
return hex.EncodeToString(hash.Sum(nil))
}
// TargetActionKind identifies a Target action independently of display labels.
@@ -299,7 +310,7 @@ type targetDetectionUpdate struct {
complete bool
}
func (a *Application) beginTargetDetection() (Snapshot, <-chan targetDetectionUpdate) {
func (a *Application) beginTargetDetection(ctx context.Context) (Snapshot, <-chan targetDetectionUpdate) {
a.stateMu.Lock()
a.detectionGeneration++
generation := a.detectionGeneration
@@ -316,7 +327,7 @@ func (a *Application) beginTargetDetection() (Snapshot, <-chan targetDetectionUp
for i, adapter := range a.targets {
adapter, id := adapter, a.targetIDs[i]
go func() {
outcomes <- targetDetectionOutcome{targetID: id, detection: detectTarget(adapter, active)}
outcomes <- targetDetectionOutcome{targetID: id, detection: detectTarget(ctx, adapter, active)}
}()
}
go func() {
@@ -351,28 +362,30 @@ func (a *Application) beginTargetDetection() (Snapshot, <-chan targetDetectionUp
return checking, updates
}
func detectTarget(adapter target.Target, active *profile.Profile) target.Detection {
func detectTarget(ctx context.Context, adapter target.Target, active *profile.Profile) target.Detection {
if detector, ok := adapter.(target.Detector); ok {
return detector.Detect(context.Background(), active)
return detector.Detect(ctx, active)
}
return target.Detection{
Availability: adapter.Availability(context.Background()),
Inspection: adapter.Inspect(context.Background(), active),
Availability: adapter.Availability(ctx),
Inspection: adapter.Inspect(ctx, active),
}
}
func (a *Application) Dispatch(intent Intent) Result {
return a.dispatch(intent, nil)
ctx, cancel := context.WithTimeout(context.Background(), operationTimeout)
defer cancel()
return a.dispatch(ctx, intent, nil)
}
func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
func (a *Application) dispatch(ctx context.Context, intent Intent, publish func(Result)) Result {
// Serialize local mutation workflows even when tests intentionally omit the
// cross-process MutationLock.
a.dispatchMu.Lock()
defer a.dispatchMu.Unlock()
if a.mutationLock != nil {
unlock, err := a.mutationLock.Lock(context.Background())
unlock, err := a.mutationLock.Lock(ctx)
if err != nil {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("lock proxyctl mutation: %w", err)}
}
@@ -388,16 +401,16 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
}
a.replaceState(fresh)
if intent.Kind == ActivateProfile {
a.refreshInspections()
a.refreshInspections(ctx)
}
if isTargetMutationIntent(intent.Kind) {
return a.setTargetDesired(intent, publish)
return a.setTargetDesired(ctx, intent, publish)
}
if intent.Kind == RefreshTargets {
return a.refreshTargets(publish)
return a.refreshTargets(ctx, publish)
}
if intent.Kind == ActivateProfile {
return a.activateProfile(intent, publish)
return a.activateProfile(ctx, intent, publish)
}
var notice string
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
@@ -416,7 +429,7 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
})
if err == nil {
a.replaceState(next)
a.refreshInspections()
a.refreshInspections(ctx)
} else {
var committed *profile.CommitError
if errors.As(err, &committed) {
@@ -434,7 +447,7 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
return Result{Snapshot: a.Snapshot(), Notice: notice, Err: err}
}
func (a *Application) activateProfile(intent Intent, publish func(Result)) Result {
func (a *Application) activateProfile(ctx context.Context, intent Intent, publish func(Result)) Result {
state := a.stateCopy()
if index(state, intent.Name) < 0 {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("Profile %q does not exist", intent.Name)}
@@ -446,23 +459,28 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
inspections := a.inspectionsCopy()
var enabled []int
var warnings []string
confirmationScope := []string{"activate-profile", intent.Name}
for i, adapter := range a.targets {
id := a.targetIDs[i]
if inspections[id].State != target.Enabled {
continue
}
enabled = append(enabled, i)
confirmationScope = append(confirmationScope, id)
if warning := adapter.Confirmation(true); warning != "" {
warnings = append(warnings, a.targetDescriptions[i].Name+": "+warning)
detail := a.targetDescriptions[i].Name + ": " + warning
warnings = append(warnings, detail)
confirmationScope = append(confirmationScope, detail)
} else {
confirmationScope = append(confirmationScope, "")
}
}
if len(warnings) > 0 && !intent.Confirm {
message := "Changing the Active Profile requires confirmation:\n" + strings.Join(warnings, "\n")
token := confirmationToken(confirmationScope...)
if len(warnings) > 0 && (!intent.Confirm || intent.ConfirmationToken != token) {
confirmation := &Confirmation{
Title: "Confirm Active Profile change", Summary: "Changing the Active Profile has high-impact Target effects.",
Details: warnings, ConfirmLabel: "Change Profile", CancelLabel: "Cancel", HighImpact: true,
Details: warnings, ConfirmLabel: "Change Profile", CancelLabel: "Cancel", HighImpact: true, Token: token,
}
_ = message
return Result{Snapshot: a.Snapshot(), ConfirmationRequest: confirmation,
Status: OperationStatus{Summary: confirmation.Title}}
}
@@ -489,7 +507,7 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
// without the state lock, then publish all observations atomically.
observed := make(map[string]target.Inspection, len(a.targets))
for i, adapter := range a.targets {
observed[a.targetIDs[i]] = adapter.Inspect(context.Background(), active)
observed[a.targetIDs[i]] = adapter.Inspect(ctx, active)
}
a.replaceInspections(observed)
@@ -505,16 +523,25 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
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)
operation := adapter.SetDesired(ctx, true, active)
inspection := operation.Inspection
var safeErr error
if operation.Err != nil {
// A failure-returned inspection may predate a partial write. Always
// inspect explicitly against the new Active Profile before publishing.
inspection = adapter.Inspect(context.Background(), active)
inspection = adapter.Inspect(ctx, active)
safeErr = fmt.Errorf("%s (fresh inspection: %s — %s)", profile.RedactText(operation.Err.Error()), inspection.State, profile.RedactText(inspection.Evidence))
}
a.replaceInspection(id, inspection)
if a.logger != nil {
outcome, detail := "success", inspection.Evidence
if safeErr != nil {
outcome, detail = "failure", safeErr.Error()
}
if logErr := a.logger.Write(LogEntry{Time: time.Now(), Operation: "switch-Active-Profile", Target: id, Result: outcome, Detail: RedactLogDetail(detail)}); logErr != nil {
safeErr = errors.Join(safeErr, fmt.Errorf("write %s Target operation log: %w", name, logErr))
}
}
progressPhase := TargetProgressSucceeded
if safeErr != nil {
progressPhase = TargetProgressFailed
@@ -528,13 +555,6 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
}
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{item}, Notice: name + " Target update completed", Status: status, Err: safeErr})
}
if a.logger != nil {
outcome, detail := "success", inspection.Evidence
if safeErr != nil {
outcome, detail = "failure", safeErr.Error()
}
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: "switch-Active-Profile", Target: id, Result: outcome, Detail: RedactLogDetail(detail)})
}
if safeErr != nil {
failures = append(failures, fmt.Errorf("%s Target: %w", name, safeErr))
}
@@ -545,7 +565,7 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
for i, adapter := range a.targets {
id := a.targetIDs[i]
if !updated[id] {
remaining[id] = adapter.Inspect(context.Background(), active)
remaining[id] = adapter.Inspect(ctx, active)
}
}
a.mergeInspections(remaining)
@@ -670,8 +690,8 @@ func targetByID(snapshot Snapshot, id string) (TargetSnapshot, string) {
return TargetSnapshot{ID: id}, id
}
func (a *Application) refreshTargets(publish func(Result)) Result {
checking, updates := a.beginTargetDetection()
func (a *Application) refreshTargets(ctx context.Context, publish func(Result)) Result {
checking, updates := a.beginTargetDetection(ctx)
if publish != nil {
progress := make([]TargetProgress, 0, len(checking.Targets))
for _, item := range checking.Targets {
@@ -714,11 +734,11 @@ func (a *Application) refreshTargets(publish func(Result)) Result {
return result
}
func (a *Application) refreshInspections() {
func (a *Application) refreshInspections(ctx context.Context) {
active := a.activeProfile()
inspections := make(map[string]target.Inspection, len(a.targets))
for i, adapter := range a.targets {
inspections[a.targetIDs[i]] = adapter.Inspect(context.Background(), active)
inspections[a.targetIDs[i]] = adapter.Inspect(ctx, active)
}
a.replaceInspections(inspections)
}
@@ -769,7 +789,7 @@ func validateTargetAction(intent Intent, selected TargetSnapshot) (targetActionP
return policy, nil
}
func (a *Application) setTargetDesired(intent Intent, publish func(Result)) Result {
func (a *Application) setTargetDesired(ctx context.Context, intent Intent, publish func(Result)) Result {
selectedIndex := -1
for i, id := range a.targetIDs {
if id == intent.TargetID {
@@ -785,27 +805,30 @@ func (a *Application) setTargetDesired(intent Intent, publish func(Result)) Resu
// This targeted check runs inside dispatch serialization and immediately
// before action validation. Adapter calls happen without stateMu; the full
// outcome is then published under one lock before policy is derived.
detection := detectTarget(selected, active)
detection := detectTarget(ctx, selected, active)
snapshot := a.publishTargetDetection(intent.TargetID, detection)
selectedSnapshot := snapshot.Targets[selectedIndex]
policy, err := validateTargetAction(intent, selectedSnapshot)
if err != nil {
return Result{Snapshot: snapshot, Err: err}
}
if warning := selected.Confirmation(policy.desired); 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(), ConfirmationRequest: confirmation,
Status: OperationStatus{Summary: confirmation.Title, TargetID: intent.TargetID}}
if warning := selected.Confirmation(policy.desired); warning != "" {
token := confirmationToken("target", intent.TargetID, fmt.Sprint(intent.Kind), fmt.Sprint(selectedSnapshot.State), fmt.Sprint(policy.desired), warning)
if !intent.Confirm || intent.ConfirmationToken != token {
confirmation := &Confirmation{Title: "Confirm " + a.targetDescriptions[selectedIndex].Name + " Target change", Summary: warning,
ConfirmLabel: "Continue", CancelLabel: "Cancel", HighImpact: true, Token: token}
return Result{Snapshot: a.Snapshot(), ConfirmationRequest: confirmation,
Status: OperationStatus{Summary: confirmation.Title, TargetID: intent.TargetID}}
}
}
if publish != nil {
publish(Result{Snapshot: snapshot, Progress: []TargetProgress{{TargetID: intent.TargetID, Phase: TargetProgressRunning, State: selectedSnapshot.State}},
Status: OperationStatus{Summary: "Updating " + a.targetDescriptions[selectedIndex].Name + " Target", TargetID: intent.TargetID}})
}
result := selected.SetDesired(context.Background(), policy.desired, active)
result := selected.SetDesired(ctx, policy.desired, active)
inspection := result.Inspection
if result.Err != nil {
inspection = selected.Inspect(context.Background(), active)
inspection = selected.Inspect(ctx, active)
}
a.replaceInspection(intent.TargetID, inspection)
outcome, detail := "success", inspection.Evidence
@@ -813,7 +836,9 @@ func (a *Application) setTargetDesired(intent Intent, publish func(Result)) Resu
outcome, detail = "failure", result.Err.Error()
}
if a.logger != nil {
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: policy.operation, Target: intent.TargetID, Result: outcome, Detail: RedactLogDetail(detail)})
if logErr := a.logger.Write(LogEntry{Time: time.Now(), Operation: policy.operation, Target: intent.TargetID, Result: outcome, Detail: RedactLogDetail(detail)}); logErr != nil {
result.Err = errors.Join(result.Err, fmt.Errorf("write Target operation log: %w", logErr))
}
}
notice := ""
progressPhase := TargetProgressSucceeded
internal/app/operation.go
+18 -3
View File
@@ -1,5 +1,12 @@
package app
import (
"context"
"time"
)
const operationTimeout = 5 * time.Minute
// StatusKind gives presentation layers semantics without requiring them to
// inspect user-facing status text.
type StatusKind uint8
@@ -27,6 +34,9 @@ type Confirmation struct {
ConfirmLabel string
CancelLabel string
HighImpact bool
// Token binds acceptance to the exact state and operation scope that was
// presented. Presentation layers must return it with the confirmed Intent.
Token string
}
// TargetProgressPhase describes one Target's part in an operation stream.
@@ -189,6 +199,7 @@ func (u OperationUpdate) IsTerminal() bool { return u.Decode().IsTerm
type OperationStream struct {
Snapshot Snapshot
Updates <-chan OperationUpdate
Cancel context.CancelFunc
}
// InteractiveModule is the presentation-independent Application seam used by
@@ -216,16 +227,20 @@ func (a *Application) StartOperation(intent Intent) OperationStream {
a.interactiveOperation = true
a.interactiveMu.Unlock()
ctx, cancel := context.WithTimeout(context.Background(), operationTimeout)
updates := make(chan OperationUpdate, 2*len(a.targets)+2)
go func() {
defer close(updates)
defer a.finishInteractiveOperation()
final := a.dispatch(intent, func(result Result) {
defer cancel()
final := a.dispatch(ctx, intent, func(result Result) {
updates <- runningOperationUpdate(result)
})
// Release the Application gate before publishing the terminal update so
// an immediate confirmation retry cannot observe a stale busy state.
a.finishInteractiveOperation()
updates <- terminalOperationUpdate(final)
}()
return OperationStream{Snapshot: initial, Updates: updates}
return OperationStream{Snapshot: initial, Updates: updates, Cancel: cancel}
}
func (a *Application) finishInteractiveOperation() {
internal/app/operation_stream_test.go
+53 -1
View File
@@ -83,6 +83,45 @@ func (t *coordinatedOperationTarget) SetDesired(_ context.Context, enabled bool,
return target.OperationResult{Inspection: target.Inspection{State: t.state, Evidence: "completed"}}
}
type cancellableOperationTarget struct{ started chan struct{} }
func (t *cancellableOperationTarget) ID() string { return "cancellable" }
func (t *cancellableOperationTarget) Description() target.Description {
return target.Description{Name: "Cancellable"}
}
func (t *cancellableOperationTarget) Availability(context.Context) target.Availability {
return target.Availability{Available: true}
}
func (t *cancellableOperationTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
return target.Inspection{State: target.Disabled, Evidence: "not changed"}
}
func (t *cancellableOperationTarget) Confirmation(bool) string { return "" }
func (t *cancellableOperationTarget) SetDesired(ctx context.Context, _ bool, _ *profile.Profile) target.OperationResult {
close(t.started)
<-ctx.Done()
return target.OperationResult{Inspection: target.Inspection{State: target.Disabled}, Err: ctx.Err()}
}
func TestInteractiveOperationCancellationReachesTarget(t *testing.T) {
adapter := &cancellableOperationTarget{started: make(chan struct{})}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
}
createAndActivate(t, application)
refreshApplicationTargets(t, application)
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
<-adapter.started
operation.Cancel()
var final app.OperationUpdate
for update := range operation.Updates {
final = update
}
if !final.IsFailed() || !errors.Is(final.Err(), context.Canceled) {
t.Fatalf("cancelled operation final update = %#v", final)
}
}
func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgress(t *testing.T) {
adapter := &coordinatedOperationTarget{id: "shell", state: target.Disabled, started: make(chan struct{}), release: make(chan struct{})}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
@@ -224,9 +263,22 @@ func TestInteractiveOperationStreamUsesStructuredConfirmation(t *testing.T) {
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
update := <-operation.Updates
confirmation := update.Confirmation()
if !update.IsConfirmation() || confirmation == nil || confirmation.Title == "" || confirmation.CancelLabel == "" {
if !update.IsConfirmation() || confirmation == nil || confirmation.Title == "" || confirmation.CancelLabel == "" || confirmation.Token == "" {
t.Fatalf("confirmation update = %#v", update)
}
// Retry immediately after the terminal confirmation, before observing the
// first stream close. The Application gate must already be released.
retry := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID(), Confirm: true, ConfirmationToken: confirmation.Token})
var final app.OperationUpdate
for next := range retry.Updates {
if next.Decode().Kind == app.OperationClosed {
t.Fatalf("immediate confirmed retry observed stale operation gate: %#v", next)
}
final = next
}
if !final.IsSucceeded() {
t.Fatalf("confirmed retry final update = %#v", final)
}
if _, open := <-operation.Updates; open {
t.Fatal("confirmation operation stream did not close")
}
internal/app/target_test.go
+25 -1
View File
@@ -92,7 +92,12 @@ func TestApplicationRequiresTargetConfirmationBeforeMutation(t *testing.T) {
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.EnableTarget, TargetID: "docker-daemon", Confirm: true})
adapter.confirmation = "Updated restart warning after state changed."
stale := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "docker-daemon", Confirm: true, ConfirmationToken: warning.ConfirmationRequest.Token})
if stale.ConfirmationRequest == nil || stale.ConfirmationRequest.Token == warning.ConfirmationRequest.Token || len(adapter.calls) != 0 {
t.Fatalf("stale confirmation was accepted = %#v, calls=%v", stale, adapter.calls)
}
confirmed := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "docker-daemon", Confirm: true, ConfirmationToken: stale.ConfirmationRequest.Token})
if confirmed.Err != nil || len(adapter.calls) != 1 || confirmed.Snapshot.Targets[0].State != target.Enabled {
t.Fatalf("confirmed result = %#v, calls=%v", confirmed, adapter.calls)
}
@@ -105,6 +110,25 @@ type targetWithID struct {
func (t targetWithID) ID() string { return t.id }
func TestTargetMutationReportsLoggingFailureAfterKeepingAppliedState(t *testing.T) {
adapter := &fakeTarget{available: true, state: target.Disabled}
logger := &recordingLog{err: errors.New("log device full")}
application, err := app.NewWithTargetsAndRuntime(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter}, nil, logger)
if err != nil {
t.Fatal(err)
}
createAndActivate(t, application)
refreshApplicationTargets(t, application)
result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
if result.Err == nil || !strings.Contains(result.Err.Error(), "operation log") || result.Snapshot.Targets[0].State != target.Enabled {
t.Fatalf("logging failure result = %#v", result)
}
if len(logger.entries) != 1 {
t.Fatalf("log attempts = %#v", logger.entries)
}
}
func TestApplicationHidesUnavailableTargetButKeepsExplanation(t *testing.T) {
adapter := &fakeTarget{available: false, state: target.Disabled}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})