fix: finish progressive Target detection
This commit is contained in:
+197
-103
@@ -53,17 +53,12 @@ type TargetSnapshot struct {
|
||||
Evidence string
|
||||
}
|
||||
|
||||
type TargetDetectionResult struct {
|
||||
TargetID string
|
||||
Phase TargetDetectionPhase
|
||||
Availability target.Availability
|
||||
Inspection target.Inspection
|
||||
Err error
|
||||
}
|
||||
|
||||
// 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
|
||||
Result TargetDetectionResult
|
||||
TargetID string
|
||||
Complete bool
|
||||
}
|
||||
|
||||
@@ -110,16 +105,20 @@ type ProgressiveDetectionModule interface {
|
||||
}
|
||||
|
||||
type Application struct {
|
||||
store profile.Store
|
||||
state profile.State
|
||||
targets []target.Target
|
||||
availabilities map[string]target.Availability
|
||||
inspections map[string]target.Inspection
|
||||
detectionPhase map[string]TargetDetectionPhase
|
||||
detectionErr map[string]error
|
||||
detectionMu sync.RWMutex
|
||||
mutationLock MutationLock
|
||||
logger OperationLogger
|
||||
store profile.Store
|
||||
state profile.State
|
||||
targets []target.Target
|
||||
targetIDs []string
|
||||
targetDescriptions []target.Description
|
||||
availabilities map[string]target.Availability
|
||||
inspections map[string]target.Inspection
|
||||
detectionPhase map[string]TargetDetectionPhase
|
||||
detectionErr map[string]error
|
||||
detectionGeneration uint64
|
||||
stateMu sync.RWMutex
|
||||
dispatchMu sync.Mutex
|
||||
mutationLock MutationLock
|
||||
logger OperationLogger
|
||||
}
|
||||
|
||||
func New(store profile.Store) (*Application, error) {
|
||||
@@ -155,73 +154,87 @@ func NewWithTargetsAndRuntime(store profile.Store, targets []target.Target, muta
|
||||
logger: logger,
|
||||
}
|
||||
for _, adapter := range a.targets {
|
||||
a.detectionPhase[adapter.ID()] = TargetChecking
|
||||
id := adapter.ID()
|
||||
a.targetIDs = append(a.targetIDs, id)
|
||||
a.targetDescriptions = append(a.targetDescriptions, adapter.Description())
|
||||
a.detectionPhase[id] = TargetChecking
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
|
||||
func (a *Application) Snapshot() Snapshot {
|
||||
a.detectionMu.RLock()
|
||||
defer a.detectionMu.RUnlock()
|
||||
a.stateMu.RLock()
|
||||
defer a.stateMu.RUnlock()
|
||||
return a.snapshotLocked()
|
||||
}
|
||||
|
||||
func (a *Application) snapshotLocked() Snapshot {
|
||||
profiles := append([]profile.Profile(nil), a.state.Profiles...)
|
||||
snapshot := Snapshot{Profiles: profiles, ActiveProfile: a.state.ActiveProfile}
|
||||
for _, adapter := range a.targets {
|
||||
id := adapter.ID()
|
||||
state := cloneState(a.state)
|
||||
snapshot := Snapshot{Profiles: state.Profiles, ActiveProfile: state.ActiveProfile}
|
||||
for i, id := range a.targetIDs {
|
||||
availability := a.availabilities[id]
|
||||
inspection := a.inspections[id]
|
||||
snapshot.Targets = append(snapshot.Targets, TargetSnapshot{
|
||||
ID: id, Description: adapter.Description(), DetectionPhase: a.detectionPhase[id], DetectionErr: a.detectionErr[id],
|
||||
ID: id, Description: a.targetDescriptions[i], DetectionPhase: a.detectionPhase[id], DetectionErr: a.detectionErr[id],
|
||||
Available: availability.Available, AvailabilityEvidence: availability.Evidence, State: inspection.State, Evidence: inspection.Evidence,
|
||||
})
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
|
||||
type targetDetectionOutcome struct {
|
||||
targetID string
|
||||
detection target.Detection
|
||||
}
|
||||
|
||||
func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate) {
|
||||
a.detectionMu.Lock()
|
||||
for _, adapter := range a.targets {
|
||||
id := adapter.ID()
|
||||
a.stateMu.Lock()
|
||||
a.detectionGeneration++
|
||||
generation := a.detectionGeneration
|
||||
for _, id := range a.targetIDs {
|
||||
a.detectionPhase[id] = TargetChecking
|
||||
a.detectionErr[id] = nil
|
||||
}
|
||||
active := activeProfile(a.state)
|
||||
checking := a.snapshotLocked()
|
||||
a.detectionMu.Unlock()
|
||||
a.stateMu.Unlock()
|
||||
|
||||
updates := make(chan TargetDetectionUpdate, len(a.targets)+1)
|
||||
active := a.activeProfile()
|
||||
outcomes := make(chan TargetDetectionResult, len(a.targets))
|
||||
for _, adapter := range a.targets {
|
||||
adapter := adapter
|
||||
outcomes := make(chan targetDetectionOutcome, len(a.targets))
|
||||
for i, adapter := range a.targets {
|
||||
adapter, id := adapter, a.targetIDs[i]
|
||||
go func() {
|
||||
detection := detectTarget(adapter, active)
|
||||
phase := TargetReady
|
||||
if detection.Err != nil {
|
||||
phase = TargetDetectionFailed
|
||||
}
|
||||
outcomes <- TargetDetectionResult{
|
||||
TargetID: adapter.ID(), Phase: phase, Availability: detection.Availability,
|
||||
Inspection: detection.Inspection, Err: detection.Err,
|
||||
}
|
||||
outcomes <- targetDetectionOutcome{targetID: id, detection: detectTarget(adapter, active)}
|
||||
}()
|
||||
}
|
||||
go func() {
|
||||
defer close(updates)
|
||||
for range a.targets {
|
||||
result := <-outcomes
|
||||
a.detectionMu.Lock()
|
||||
a.availabilities[result.TargetID] = result.Availability
|
||||
a.inspections[result.TargetID] = result.Inspection
|
||||
a.detectionPhase[result.TargetID] = result.Phase
|
||||
a.detectionErr[result.TargetID] = result.Err
|
||||
snapshot := a.snapshotLocked()
|
||||
a.detectionMu.Unlock()
|
||||
updates <- TargetDetectionUpdate{Snapshot: snapshot, Result: result}
|
||||
outcome := <-outcomes
|
||||
a.stateMu.Lock()
|
||||
if generation != a.detectionGeneration {
|
||||
a.stateMu.Unlock()
|
||||
continue
|
||||
}
|
||||
phase := TargetReady
|
||||
if outcome.detection.Err != nil {
|
||||
phase = TargetDetectionFailed
|
||||
}
|
||||
a.availabilities[outcome.targetID] = outcome.detection.Availability
|
||||
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}
|
||||
// The buffered send cannot wait on a consumer and happens before a
|
||||
// newer Begin call can make this generation stale.
|
||||
updates <- update
|
||||
a.stateMu.Unlock()
|
||||
}
|
||||
updates <- TargetDetectionUpdate{Snapshot: a.Snapshot(), Complete: true}
|
||||
a.stateMu.Lock()
|
||||
if generation == a.detectionGeneration {
|
||||
updates <- TargetDetectionUpdate{Snapshot: a.snapshotLocked(), Complete: true}
|
||||
}
|
||||
a.stateMu.Unlock()
|
||||
}()
|
||||
return checking, updates
|
||||
}
|
||||
@@ -252,6 +265,11 @@ func (a *Application) DispatchProgress(intent Intent) <-chan Result {
|
||||
}
|
||||
|
||||
func (a *Application) dispatch(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())
|
||||
if err != nil {
|
||||
@@ -267,7 +285,7 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
|
||||
if err != nil {
|
||||
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("reload Profile state for mutation: %w", err)}
|
||||
}
|
||||
a.state = fresh
|
||||
a.replaceState(fresh)
|
||||
if intent.Kind == SetTargetDesired || intent.Kind == ActivateProfile {
|
||||
a.refreshInspections()
|
||||
}
|
||||
@@ -297,19 +315,19 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
|
||||
return current, mutationErr
|
||||
})
|
||||
if err == nil {
|
||||
a.state = next
|
||||
a.replaceState(next)
|
||||
a.refreshInspections()
|
||||
} else {
|
||||
var committed *profile.CommitError
|
||||
if errors.As(err, &committed) {
|
||||
// Rename committed next before directory durability reporting failed.
|
||||
// Prefer that state even if the required verification reload also fails.
|
||||
a.state = next
|
||||
a.replaceState(next)
|
||||
reloaded, loadErr := a.store.Load()
|
||||
if loadErr != nil {
|
||||
err = errors.Join(err, fmt.Errorf("reload committed Profile configuration: %w", loadErr))
|
||||
} else {
|
||||
a.state = reloaded
|
||||
a.replaceState(reloaded)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -317,22 +335,25 @@ func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
|
||||
}
|
||||
|
||||
func (a *Application) activateProfile(intent Intent, publish func(Result)) Result {
|
||||
if index(a.state, intent.Name) < 0 {
|
||||
state := a.stateCopy()
|
||||
if index(state, intent.Name) < 0 {
|
||||
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("Profile %q does not exist", intent.Name)}
|
||||
}
|
||||
if a.state.ActiveProfile == intent.Name {
|
||||
if state.ActiveProfile == intent.Name {
|
||||
return Result{Snapshot: a.Snapshot(), Notice: fmt.Sprintf("Profile %q is already Active", intent.Name)}
|
||||
}
|
||||
|
||||
var enabled []target.Target
|
||||
inspections := a.inspectionsCopy()
|
||||
var enabled []int
|
||||
var warnings []string
|
||||
for _, adapter := range a.targets {
|
||||
if a.inspections[adapter.ID()].State != target.Enabled {
|
||||
for i, adapter := range a.targets {
|
||||
id := a.targetIDs[i]
|
||||
if inspections[id].State != target.Enabled {
|
||||
continue
|
||||
}
|
||||
enabled = append(enabled, adapter)
|
||||
enabled = append(enabled, i)
|
||||
if warning := adapter.Confirmation(true); warning != "" {
|
||||
warnings = append(warnings, adapter.Description().Name+": "+warning)
|
||||
warnings = append(warnings, a.targetDescriptions[i].Name+": "+warning)
|
||||
}
|
||||
}
|
||||
if len(warnings) > 0 && !intent.Confirm {
|
||||
@@ -346,28 +367,35 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
|
||||
if err != nil {
|
||||
var committed *profile.CommitError
|
||||
if errors.As(err, &committed) {
|
||||
a.state = next
|
||||
a.replaceState(next)
|
||||
if reloaded, loadErr := a.store.Load(); loadErr != nil {
|
||||
err = errors.Join(err, fmt.Errorf("reload committed Profile configuration: %w", loadErr))
|
||||
} else {
|
||||
a.state = reloaded
|
||||
a.replaceState(reloaded)
|
||||
}
|
||||
}
|
||||
return Result{Snapshot: a.Snapshot(), Err: err}
|
||||
}
|
||||
a.state = next
|
||||
a.replaceState(next)
|
||||
active := a.activeProfile()
|
||||
// The Active Profile has changed before any Target write begins. Re-inspect
|
||||
// immediately so rows not yet updated are truthfully Drifted rather than
|
||||
// appearing Enabled against stale managed values.
|
||||
for _, adapter := range a.targets {
|
||||
a.inspections[adapter.ID()] = adapter.Inspect(context.Background(), active)
|
||||
// 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)
|
||||
}
|
||||
a.replaceInspections(observed)
|
||||
|
||||
var progress []TargetProgress
|
||||
var failures []error
|
||||
for _, adapter := range enabled {
|
||||
updated := make(map[string]bool, len(enabled))
|
||||
for _, targetIndex := range enabled {
|
||||
adapter := a.targets[targetIndex]
|
||||
id := a.targetIDs[targetIndex]
|
||||
name := a.targetDescriptions[targetIndex].Name
|
||||
updated[id] = true
|
||||
if publish != nil {
|
||||
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: adapter.ID(), Applying: true, State: a.inspections[adapter.ID()].State}}, Notice: "Updating " + adapter.Description().Name + " Target"})
|
||||
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: id, Applying: true, State: a.inspection(id).State}}, Notice: "Updating " + name + " Target"})
|
||||
}
|
||||
operation := adapter.SetDesired(context.Background(), true, active)
|
||||
inspection := operation.Inspection
|
||||
@@ -378,34 +406,33 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
|
||||
inspection = adapter.Inspect(context.Background(), active)
|
||||
safeErr = fmt.Errorf("%s (fresh inspection: %s — %s)", profile.RedactText(operation.Err.Error()), inspection.State, profile.RedactText(inspection.Evidence))
|
||||
}
|
||||
a.inspections[adapter.ID()] = inspection
|
||||
item := TargetProgress{TargetID: adapter.ID(), State: inspection.State, Evidence: inspection.Evidence, Err: safeErr}
|
||||
a.replaceInspection(id, inspection)
|
||||
item := TargetProgress{TargetID: id, State: inspection.State, Evidence: inspection.Evidence, Err: safeErr}
|
||||
progress = append(progress, item)
|
||||
if publish != nil {
|
||||
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{item}, Notice: adapter.Description().Name + " Target update completed", Err: safeErr})
|
||||
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{item}, Notice: name + " Target update completed", 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: adapter.ID(), Result: outcome, Detail: RedactLogDetail(detail)})
|
||||
_ = 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", adapter.Description().Name, safeErr))
|
||||
failures = append(failures, fmt.Errorf("%s Target: %w", name, safeErr))
|
||||
}
|
||||
}
|
||||
// Re-inspect Targets which were not part of the transition so any managed
|
||||
// configuration is interpreted against the new Active Profile as well.
|
||||
updated := make(map[string]bool, len(enabled))
|
||||
for _, adapter := range enabled {
|
||||
updated[adapter.ID()] = true
|
||||
}
|
||||
for _, adapter := range a.targets {
|
||||
if !updated[adapter.ID()] {
|
||||
a.inspections[adapter.ID()] = adapter.Inspect(context.Background(), active)
|
||||
remaining := make(map[string]target.Inspection)
|
||||
for i, adapter := range a.targets {
|
||||
id := a.targetIDs[i]
|
||||
if !updated[id] {
|
||||
remaining[id] = adapter.Inspect(context.Background(), active)
|
||||
}
|
||||
}
|
||||
a.mergeInspections(remaining)
|
||||
result := Result{Snapshot: a.Snapshot(), Progress: progress, Notice: fmt.Sprintf("Activated Profile %q", intent.Name)}
|
||||
if len(failures) > 0 {
|
||||
result.Err = errors.Join(failures...)
|
||||
@@ -414,16 +441,80 @@ func (a *Application) activateProfile(intent Intent, publish func(Result)) Resul
|
||||
return result
|
||||
}
|
||||
|
||||
func (a *Application) activeProfile() *profile.Profile {
|
||||
for i := range a.state.Profiles {
|
||||
if a.state.Profiles[i].Name == a.state.ActiveProfile {
|
||||
value := a.state.Profiles[i]
|
||||
func activeProfile(state profile.State) *profile.Profile {
|
||||
for i := range state.Profiles {
|
||||
if state.Profiles[i].Name == state.ActiveProfile {
|
||||
value := state.Profiles[i]
|
||||
value.Bypass = append([]string(nil), value.Bypass...)
|
||||
return &value
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Application) activeProfile() *profile.Profile {
|
||||
a.stateMu.RLock()
|
||||
defer a.stateMu.RUnlock()
|
||||
return activeProfile(a.state)
|
||||
}
|
||||
|
||||
func (a *Application) stateCopy() profile.State {
|
||||
a.stateMu.RLock()
|
||||
defer a.stateMu.RUnlock()
|
||||
return cloneState(a.state)
|
||||
}
|
||||
|
||||
func (a *Application) replaceState(state profile.State) {
|
||||
a.stateMu.Lock()
|
||||
defer a.stateMu.Unlock()
|
||||
a.detectionGeneration++
|
||||
a.state = cloneState(state)
|
||||
}
|
||||
|
||||
func (a *Application) inspectionsCopy() map[string]target.Inspection {
|
||||
a.stateMu.RLock()
|
||||
defer a.stateMu.RUnlock()
|
||||
result := make(map[string]target.Inspection, len(a.inspections))
|
||||
for id, inspection := range a.inspections {
|
||||
result[id] = inspection
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func (a *Application) inspection(id string) target.Inspection {
|
||||
a.stateMu.RLock()
|
||||
defer a.stateMu.RUnlock()
|
||||
return a.inspections[id]
|
||||
}
|
||||
|
||||
func (a *Application) replaceInspection(id string, inspection target.Inspection) {
|
||||
a.stateMu.Lock()
|
||||
defer a.stateMu.Unlock()
|
||||
a.detectionGeneration++
|
||||
a.inspections[id] = inspection
|
||||
}
|
||||
|
||||
func (a *Application) replaceInspections(inspections map[string]target.Inspection) {
|
||||
a.stateMu.Lock()
|
||||
defer a.stateMu.Unlock()
|
||||
a.detectionGeneration++
|
||||
for _, id := range a.targetIDs {
|
||||
a.inspections[id] = inspections[id]
|
||||
}
|
||||
}
|
||||
|
||||
func (a *Application) mergeInspections(inspections map[string]target.Inspection) {
|
||||
if len(inspections) == 0 {
|
||||
return
|
||||
}
|
||||
a.stateMu.Lock()
|
||||
defer a.stateMu.Unlock()
|
||||
a.detectionGeneration++
|
||||
for id, inspection := range inspections {
|
||||
a.inspections[id] = inspection
|
||||
}
|
||||
}
|
||||
|
||||
func (a *Application) refreshTargets() {
|
||||
_, updates := a.BeginTargetDetection()
|
||||
for range updates {
|
||||
@@ -432,25 +523,25 @@ func (a *Application) refreshTargets() {
|
||||
|
||||
func (a *Application) refreshInspections() {
|
||||
active := a.activeProfile()
|
||||
for _, adapter := range a.targets {
|
||||
inspection := adapter.Inspect(context.Background(), active)
|
||||
a.detectionMu.Lock()
|
||||
a.inspections[adapter.ID()] = inspection
|
||||
a.detectionMu.Unlock()
|
||||
inspections := make(map[string]target.Inspection, len(a.targets))
|
||||
for i, adapter := range a.targets {
|
||||
inspections[a.targetIDs[i]] = adapter.Inspect(context.Background(), active)
|
||||
}
|
||||
a.replaceInspections(inspections)
|
||||
}
|
||||
|
||||
func (a *Application) setTargetDesired(intent Intent) Result {
|
||||
var selected target.Target
|
||||
for _, adapter := range a.targets {
|
||||
if adapter.ID() == intent.TargetID {
|
||||
selected = adapter
|
||||
selectedIndex := -1
|
||||
for i, id := range a.targetIDs {
|
||||
if id == intent.TargetID {
|
||||
selectedIndex = i
|
||||
break
|
||||
}
|
||||
}
|
||||
if selected == nil {
|
||||
if selectedIndex < 0 {
|
||||
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("Target %q does not exist", intent.TargetID)}
|
||||
}
|
||||
selected := a.targets[selectedIndex]
|
||||
active := a.activeProfile()
|
||||
if intent.Enabled && active == nil {
|
||||
return Result{Snapshot: a.Snapshot(), Err: errors.New("no Active Profile; open Profile management before enabling a Target")}
|
||||
@@ -463,7 +554,7 @@ func (a *Application) setTargetDesired(intent Intent) Result {
|
||||
if result.Err != nil {
|
||||
inspection = selected.Inspect(context.Background(), active)
|
||||
}
|
||||
a.inspections[selected.ID()] = inspection
|
||||
a.replaceInspection(intent.TargetID, inspection)
|
||||
outcome, detail := "success", inspection.Evidence
|
||||
if result.Err != nil {
|
||||
outcome, detail = "failure", result.Err.Error()
|
||||
@@ -473,11 +564,11 @@ func (a *Application) setTargetDesired(intent Intent) Result {
|
||||
if intent.Enabled {
|
||||
operation = "enable"
|
||||
}
|
||||
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: operation, Target: selected.ID(), Result: outcome, Detail: RedactLogDetail(detail)})
|
||||
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: operation, Target: intent.TargetID, Result: outcome, Detail: RedactLogDetail(detail)})
|
||||
}
|
||||
notice := ""
|
||||
if result.Err == nil {
|
||||
notice = fmt.Sprintf("%s Target is now %s", selected.Description().Name, inspection.State)
|
||||
notice = fmt.Sprintf("%s Target is now %s", a.targetDescriptions[selectedIndex].Name, inspection.State)
|
||||
}
|
||||
return Result{Snapshot: a.Snapshot(), Notice: notice, Err: result.Err}
|
||||
}
|
||||
@@ -551,5 +642,8 @@ func index(state profile.State, name string) int {
|
||||
|
||||
func cloneState(state profile.State) profile.State {
|
||||
state.Profiles = append([]profile.Profile(nil), state.Profiles...)
|
||||
for i := range state.Profiles {
|
||||
state.Profiles[i].Bypass = append([]string(nil), state.Profiles[i].Bypass...)
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user