From 7862e7176c9a75ba90fb9fc16e7b401a801618db Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Mon, 17 Aug 2026 13:27:34 -0700 Subject: [PATCH 1/3] test(blackbox): compute drift end states deterministically instead of tolerating a sync window The chaos suite tracked out-of-band changes to managed resources in a ManagedDriftedResources record and taught the invariants to skip any slot with pending drift. The record was retired at several sites before the divergence was actually reconciled, so assertions ran with the exemption gone and failed on exactly the resources the suite had drifted; keyed by native id, a second drift op on one resource also aliased onto the first op's snapshot. The suite never needed the exemption: every sync is harness-triggered (background synchronization is disabled), so drift can be absorbed synchronously at the drift operation itself and the model can compute the absorbed end state directly from generated inputs. - OOB modify/delete of a managed resource now targets only resources whose stacks have no in-flight commands, shapes the drifted properties to the slot's schema (keys outside it would split into ReadOnlyProperties), forces a sync, and waits for the inventory row to converge before continuing; the model records the absorbed state at the operation - OOB modify/delete of a discovered unmanaged resource absorbs the same way, closing the same window on the unmanaged path - delete the drift-tolerance mechanism and the invariant skips wholesale; CheckInvariants loses its managed-drift ignore set and CheckManagedDriftVsInventory is gone - capture the sync-command baseline before ForceSync: taking it after the trigger could race the created command and made awaited syncs degrade into silent no-ops - apply the model's own predicted unmanaged sync transition at every observed sync, instead of reading the sync command's updates back into the model - drop the cloud-state mirror entry once a managed modify is absorbed so a crash restart cannot resurrect stale properties over later plugin writes - enumerate drift candidates in stable slot order so the generated sequence number selects the same resource on every run --- tests/blackbox/executor.go | 313 +++++++++++---------- tests/blackbox/harness.go | 56 ++-- tests/blackbox/invariants.go | 75 +---- tests/blackbox/invariants_rename_test.go | 4 +- tests/blackbox/property_test.go | 2 +- tests/blackbox/state_model.go | 336 ++++------------------- tests/blackbox/state_model_test.go | 128 ++++++++- 7 files changed, 367 insertions(+), 547 deletions(-) diff --git a/tests/blackbox/executor.go b/tests/blackbox/executor.go index 69e1b09a0..66825d570 100644 --- a/tests/blackbox/executor.go +++ b/tests/blackbox/executor.go @@ -346,7 +346,6 @@ func (h *TestHarness) reconcileCompletedAcceptedCommands(t *testing.T, model *St cc := completed[i] t.Logf("reconcileCompletedAcceptedCommands: command %s completed early (state=%s)", cc.ac.CommandID, cc.cmd.State) correctModelFromCommandOutcome(t, &cc.cmd, model, model.Pool, cc.ac.Snapshots, corrected, cc.ac.IsReconcile) - h.reconcileManagedDriftOverriddenByCommand(t, model, &cc.cmd) } model.AcceptedCommands = remaining @@ -389,11 +388,7 @@ func (h *TestHarness) AssertAllInvariants(t *testing.T, model ...*StateModel) { // ResetAgentState, creating cloud resources that haven't been persisted // to inventory yet. When orphans are found, we clean them up and // re-check. If the orphan persists across retries, it's a real bug. - var ignoreManagedDriftNativeIDs map[string]bool - if len(model) > 0 && model[0] != nil { - ignoreManagedDriftNativeIDs = model[0].ManagedDriftNativeIDs() - } - resourceViolations := h.checkResourceInvariantsWithRetry(t, ignoreNativeIDs, ignoreManagedDriftNativeIDs) + resourceViolations := h.checkResourceInvariantsWithRetry(t, ignoreNativeIDs) violations = append(violations, resourceViolations...) if opLog, err := h.TryGetOperationLog(); err == nil { violations = append(violations, CheckOperationLogInvariants(opLog)...) @@ -412,7 +407,6 @@ func (h *TestHarness) AssertAllInvariants(t *testing.T, model ...*StateModel) { } modelViolations := CheckModelVsInventory(model[0], inventory) modelViolations = append(modelViolations, CheckUnmanagedModelVsInventory(model[0], unmanagedInventory)...) - modelViolations = append(modelViolations, CheckManagedDriftVsInventory(model[0], inventory)...) if len(modelViolations) > 0 { t.Logf("MODEL MISMATCH DEBUG: inventory has %d resources", len(inventory)) for _, res := range inventory { @@ -436,7 +430,7 @@ func (h *TestHarness) AssertAllInvariants(t *testing.T, model ...*StateModel) { // prior rapid iterations can create cloud resources after ResetAgentState; // these resolve themselves once the stale operations complete and the cloud // entries are cleaned. Genuine invariant bugs persist across retries. -func (h *TestHarness) checkResourceInvariantsWithRetry(t *testing.T, ignoreNativeIDs map[string]bool, ignoreManagedDriftNativeIDs map[string]bool) []Violation { +func (h *TestHarness) checkResourceInvariantsWithRetry(t *testing.T, ignoreNativeIDs map[string]bool) []Violation { t.Helper() const maxRetries = 3 @@ -459,7 +453,7 @@ func (h *TestHarness) checkResourceInvariantsWithRetry(t *testing.T, ignoreNativ h.RestartAgent(t, 30*time.Second) continue } - resourceViolations := CheckInvariants(inventory, cloudState, ignoreNativeIDs, ignoreManagedDriftNativeIDs) + resourceViolations := CheckInvariants(inventory, cloudState, ignoreNativeIDs) if len(resourceViolations) == 0 { return nil @@ -1044,7 +1038,7 @@ func (h *TestHarness) executeCrashAgent(t *testing.T, model *StateModel) { t.Logf(">>> OpCrashAgent: killing agent") h.KillAgent(t) - h.RestartAgent(t, 30*time.Second, model) + h.RestartAgent(t, 30*time.Second) t.Logf(">>> OpCrashAgent: agent restarted, state re-injected") @@ -1071,9 +1065,6 @@ func (h *TestHarness) executeCrashAgent(t *testing.T, model *StateModel) { h.DrainPendingCommands(t, model, 60*time.Second) } - // Pending managed drift stays pending after crash. The assertion skips - // slots with pending drift. Sync will resolve it in a future iteration. - // With StackExpirer disabled, TTL only fires via ForceCheckTTL. // After crash, check if any TTL stacks are now expired and process them. h.ForceCheckTTLAndWait(t, model) @@ -1570,36 +1561,53 @@ func filterExistingResources(ids []int, stackIndex int, model *StateModel) []int func (h *TestHarness) executeTriggerSync(t *testing.T, model *StateModel) { t.Helper() - // Fire-and-forget: sync runs concurrently with user commands. Resources - // in active changesets are excluded from sync (registered upfront in the - // ChangesetExecutor), so sync only touches idle resources. - err := h.client.ForceSync() - if err != nil { - t.Logf("TriggerSync error (may be expected): %v", err) - return - } - if cmd, ok := h.WaitForNextSyncCommand(2 * time.Second); !ok { - t.Logf("TriggerSync: no new sync command observed") - return - } else if model != nil { - model.ApplySyncCommand(cmd) + // Sync runs concurrently with user commands. Resources in active + // changesets are excluded from sync (registered upfront in the + // ChangesetExecutor), so sync only touches idle resources. Managed drift + // is always absorbed at the drift operation itself, so for managed + // resources this sync is a no-op; unmanaged rows absorb their cloud state. + // + // An unobserved sync is tolerable here: with no drift pending by + // construction, the sync's model transition is a no-op either way. + if h.forceSyncAndAwait(t, model) { + t.Logf("TriggerSync: fired") + } else { + t.Logf("TriggerSync: no sync command observed (nothing to synchronize)") } - t.Logf("TriggerSync: fired") } // TriggerSyncAndWait fires a sync and waits for it to complete. Used before // final invariant checks to ensure inventory reflects cloud state after chaos // operations (cancel, crash, TTL) that can leave stale inventory entries. -func (h *TestHarness) TriggerSyncAndWait(t *testing.T) { +// As in executeTriggerSync, an unobserved sync is a model no-op. +func (h *TestHarness) TriggerSyncAndWait(t *testing.T, model *StateModel) { t.Helper() - err := h.client.ForceSync() - if err != nil { - t.Logf("TriggerSyncAndWait: error: %v", err) - return + if !h.forceSyncAndAwait(t, model) { + t.Logf("TriggerSyncAndWait: no sync command observed (nothing to synchronize)") } - if _, ok := h.WaitForNextSyncCommand(10 * time.Second); !ok { - t.Logf("TriggerSyncAndWait: no sync command observed") +} + +// forceSyncAndAwait triggers a sync and waits for its command to reach a +// terminal state. When observed, it applies the model's own predicted sync +// transition for unmanaged rows (sync reads every idle inventory row, so +// out-of-band changes to discovered unmanaged resources absorb at each sync) +// and returns true. It returns false when no sync command appeared: the +// synchronizer creates none when it finds nothing to synchronize (empty +// inventory, or every row filtered by an in-flight exclusion or a transient +// plugin-availability miss). The baseline is captured before the trigger so +// the triggered command cannot be mistaken for a pre-existing one. +func (h *TestHarness) forceSyncAndAwait(t *testing.T, model *StateModel) bool { + t.Helper() + baseline := h.SyncCommandBaseline() + require.NoError(t, h.client.ForceSync(), "ForceSync failed") + _, ok := h.WaitForSyncCommandAfter(baseline, 15*time.Second) + if !ok { + return false + } + if model != nil { + model.ApplySyncToUnmanaged() } + return true } func (h *TestHarness) executeTriggerDiscovery(t *testing.T, model *StateModel) { @@ -1629,8 +1637,14 @@ func (h *TestHarness) executeCloudModify(t *testing.T, op *Operation, model *Sta t.Logf("[op %d] CloudModify: %s → skipped (resource does not exist in cloud)", op.SequenceNum, op.NativeID) return } + inInventory := res.PresentInInventory h.putCloudStateWithRetry(t, op.NativeID, res.ResourceType, op.Properties) model.ApplyUnmanagedCloudModify(op.NativeID, op.Properties) + // A discovered resource has an inventory row that sync reads, so the + // change must be absorbed before the next inventory-vs-cloud check. + if inInventory { + h.absorbUnmanagedDrift(t, model, op.NativeID, op.Properties, false) + } t.Logf("[op %d] CloudModify: %s", op.SequenceNum, op.NativeID) } @@ -1640,64 +1654,156 @@ func (h *TestHarness) executeCloudDelete(t *testing.T, op *Operation, model *Sta h.executeManagedCloudDelete(t, op, model) return } + res := model.UnmanagedResources[op.NativeID] + inInventory := res != nil && res.PresentInInventory h.deleteCloudStateWithRetry(t, op.NativeID) model.ApplyUnmanagedCloudDelete(op.NativeID) + // A discovered resource has an inventory row; sync observes the deletion + // and drops the row, so absorb it before the next model-vs-inventory check. + if inInventory { + h.absorbUnmanagedDrift(t, model, op.NativeID, "", true) + } t.Logf("[op %d] CloudDelete: %s", op.SequenceNum, op.NativeID) } +// executeManagedCloudModify injects out-of-band drift into a managed +// resource's cloud state and absorbs it immediately: it forces a sync, waits +// for the inventory row to converge on the drifted properties, and records +// that end state in the model. Absorption is deterministic because the target +// is untouched by in-flight commands (FindDriftEligibleResource) and every +// sync is harness-triggered. func (h *TestHarness) executeManagedCloudModify(t *testing.T, op *Operation, model *StateModel) { t.Helper() - stackIdx, slotIdx, stackLabel, resLabel, resType, nativeID, ok := model.FindExistingResourceWithNativeID(op.SequenceNum) + stackIdx, slotIdx, _, resLabel, resType, nativeID, ok := model.FindDriftEligibleResource(op.SequenceNum) if !ok { t.Logf("[op %d] CloudModify managed → skipped (no eligible resource in model)", op.SequenceNum) return } - _ = stackIdx - _ = slotIdx - h.putCloudStateWithRetry(t, nativeID, resType, op.Properties) - model.ApplyManagedCloudModify(stackLabel, resLabel, resType, nativeID, op.Properties) - h.reconcileManagedDriftBeforeNextCommands(t, model) + props := driftPropertiesForSlot(model, stackIdx, slotIdx, op.Properties) + h.putCloudStateWithRetry(t, nativeID, resType, props) + model.ApplyManagedCloudModify(stackIdx, slotIdx, props) + h.absorbManagedDrift(t, model, nativeID, props, false) + // Inventory now carries the absorbed properties and is the restart + // re-injection source for this resource; a lingering mirror entry would + // overwrite later plugin writes with these stale properties after a crash. + delete(h.cloudStateMirror, nativeID) t.Logf("[op %d] CloudModify managed: %s (%s)", op.SequenceNum, resLabel, nativeID) } +// executeManagedCloudDelete injects an out-of-band delete of a managed +// resource's cloud state and absorbs it immediately: the sync observes the +// deletion and drops the resource from inventory, which the model records as +// the end state. func (h *TestHarness) executeManagedCloudDelete(t *testing.T, op *Operation, model *StateModel) { t.Helper() - stackIdx, slotIdx, stackLabel, resLabel, resType, nativeID, ok := model.FindExistingResourceWithNativeID(op.SequenceNum) + stackIdx, slotIdx, _, resLabel, _, nativeID, ok := model.FindDriftEligibleResource(op.SequenceNum) if !ok { t.Logf("[op %d] CloudDelete managed → skipped (no eligible resource in model)", op.SequenceNum) return } - _ = stackIdx - _ = slotIdx h.deleteCloudStateWithRetry(t, nativeID) - model.ApplyManagedCloudDelete(stackLabel, resLabel, resType, nativeID) - h.reconcileManagedDriftBeforeNextCommands(t, model) + model.ApplyManagedCloudDelete(stackIdx, slotIdx) + h.absorbManagedDrift(t, model, nativeID, "", true) t.Logf("[op %d] CloudDelete managed: %s (%s)", op.SequenceNum, resLabel, nativeID) } -func (h *TestHarness) reconcileManagedDriftBeforeNextCommands(t *testing.T, model *StateModel) { - t.Helper() - if model == nil || len(model.ManagedDriftedResources) == 0 { - return +// driftPropertiesForSlot shapes drawn out-of-band properties to the slot's +// schema so sync absorption converges the inventory row onto exactly these +// values. Keys outside the slot's schema would split into ReadOnlyProperties +// and diverge inventory from cloud state, so child/grandchild/cross-stack +// slots get the child shape with ParentId pointing at the current parent +// identifier. +func driftPropertiesForSlot(model *StateModel, stackIdx, slotIdx int, drawnProps string) string { + if model.Pool == nil || model.Pool.IsParent(slotIdx) { + return drawnProps + } + var drawn map[string]any + if err := json.Unmarshal([]byte(drawnProps), &drawn); err != nil { + return drawnProps + } + props := map[string]any{ + "Name": drawn["Name"], + "ParentId": model.parentIdentifierForResource(stackIdx, slotIdx), + "Value": drawn["Value"], + } + b, err := json.Marshal(props) + if err != nil { + return drawnProps } - // Pending managed drift is an expected divergence until an explicit sync. - // Before subsequent command submissions, restore the main stack/slot model to - // its pre-drift snapshot so command snapshots remain based on the agent's - // current inventory rather than the unsynced cloud state. - for _, drift := range model.ManagedDriftedResources { - if !drift.PendingSync { + return string(b) +} + +// absorbManagedDrift forces syncs until the inventory row for nativeID reaches +// the absorbed end state: gone when deleted, or property-converged when +// modified. A single sync can miss the target when its read consumes a stale +// programmed response left by an earlier canceled or failed command, so the +// sync is retried; the retry's read then gets the plugin's default behavior. +func (h *TestHarness) absorbManagedDrift(t *testing.T, model *StateModel, nativeID, expectedProps string, deleted bool) { + t.Helper() + const maxSyncAttempts = 3 + for attempt := range maxSyncAttempts { + if !h.forceSyncAndAwait(t, model) { + // The drifted row is in inventory, so a healthy sync must include + // it; an unobserved sync command here is a transient miss — retry. + t.Logf("absorbManagedDrift: no sync command observed (attempt %d)", attempt+1) continue } - stackIdx, slotIdx, ok := model.findResourceSlot(drift.StackLabel, drift.ResourceLabel) - if !ok { + if h.waitForAbsorbedInventory(t, "managed:true", nativeID, expectedProps, deleted, 10*time.Second) { + return + } + t.Logf("absorbManagedDrift: %s not absorbed after sync (attempt %d)", nativeID, attempt+1) + } + require.Failf(t, "managed drift not absorbed", + "resource %s did not reach its absorbed state (deleted=%v) after %d syncs", nativeID, deleted, maxSyncAttempts) +} + +// absorbUnmanagedDrift forces syncs until the unmanaged inventory row for +// nativeID reaches its absorbed end state, mirroring absorbManagedDrift. +func (h *TestHarness) absorbUnmanagedDrift(t *testing.T, model *StateModel, nativeID, expectedProps string, deleted bool) { + t.Helper() + const maxSyncAttempts = 3 + for attempt := range maxSyncAttempts { + if !h.forceSyncAndAwait(t, model) { + t.Logf("absorbUnmanagedDrift: no sync command observed (attempt %d)", attempt+1) continue } - if drift.SnapshotState == StateExists { - model.ApplyCreated(stackIdx, []int{slotIdx}, drift.SnapshotProperties) - } else { - model.ApplyDestroyed(stackIdx, []int{slotIdx}) + if h.waitForAbsorbedInventory(t, "managed:false", nativeID, expectedProps, deleted, 10*time.Second) { + return + } + t.Logf("absorbUnmanagedDrift: %s not absorbed after sync (attempt %d)", nativeID, attempt+1) + } + require.Failf(t, "unmanaged drift not absorbed", + "unmanaged resource %s did not reach its absorbed state (deleted=%v) after %d syncs", nativeID, deleted, maxSyncAttempts) +} + +// waitForAbsorbedInventory polls the inventory selected by query until the row +// for nativeID is gone (deleted=true) or its properties json-equal +// expectedProps. +func (h *TestHarness) waitForAbsorbedInventory(t *testing.T, query, nativeID, expectedProps string, deleted bool, timeout time.Duration) bool { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + forma, err := h.client.ExtractResources(query) + if err == nil { + var row *pkgmodel.Resource + if forma != nil { + for i := range forma.Resources { + if forma.Resources[i].NativeID == nativeID { + row = &forma.Resources[i] + break + } + } + } + if deleted && row == nil { + return true + } + if !deleted && row != nil && jsonEqual(string(row.Properties), expectedProps) { + return true + } } + time.Sleep(100 * time.Millisecond) } + return false } func (h *TestHarness) executeCloudCreate(t *testing.T, op *Operation, model *StateModel) { @@ -1942,7 +2048,6 @@ func (h *TestHarness) DrainPendingCommands(t *testing.T, model *StateModel, time dc := drained[i] if dc.cmd != nil { correctModelFromCommandOutcome(t, dc.cmd, model, model.Pool, dc.ac.Snapshots, corrected, dc.ac.IsReconcile) - h.reconcileManagedDriftOverriddenByCommand(t, model, dc.cmd) } } h.reconcileAmbiguousFailedCommands(t, model, drained) @@ -1957,90 +2062,6 @@ func (h *TestHarness) DrainPendingCommands(t *testing.T, model *StateModel, time // After all commands are drained, any TTL that was blocked by active // commands is now unblocked. Do one final check to catch it. h.ForceCheckTTLAndWait(t, model) - - // If OOB changes were made during this iteration, fire a sync to let the - // agent reconcile them. The sync command's outcome updates the model via - // ApplySyncCommand. Any remaining unresolved drift stays pending — the - // assertion skips slots with pending drift. - if len(model.ManagedDriftedResources) > 0 { - h.executeTriggerSync(t, model) - // Restore cloud state for any drift the sync resolved (the plugin's - // Update/Delete already updated cloud state via the sync command). - // For unresolved drift, restore cloud state to match inventory so - // CheckInvariants doesn't see phantoms. - for nativeID, drift := range model.ManagedDriftedResources { - if !drift.PendingSync { - continue // already resolved by sync - } - if drift.PresentInCloud { - // OOB modify: restore cloud state to pre-drift properties - if drift.SnapshotProperties != "" { - h.TryPutCloudState(nativeID, drift.ResourceType, flattenPropertiesForCloud(json.RawMessage(drift.SnapshotProperties))) - } - } else { - // OOB delete: the resource is still in inventory (sync hasn't - // deleted it). Cloud state was already deleted by CloudDelete. - // Restore it so CheckInvariants doesn't see a phantom. - stackIdx, slotIdx, ok := model.findResourceSlot(drift.StackLabel, drift.ResourceLabel) - if ok { - res := model.Resource(stackIdx, slotIdx) - if res != nil && res.Properties != "" { - h.TryPutCloudState(nativeID, drift.ResourceType, flattenPropertiesForCloud(json.RawMessage(res.Properties))) - } - } - } - } - } -} - -func (h *TestHarness) reconcileManagedDriftOverriddenByCommand(t *testing.T, model *StateModel, cmd *apimodel.Command) { - t.Helper() - if model == nil || cmd == nil { - return - } - for _, ru := range cmd.ResourceUpdates { - if ru.State != "Success" { - continue - } - if ru.Operation == "delete" { - // Use the drift entry's NativeID (or the command response NativeID) - // to clean up cloud state. - nativeID := ru.NativeID - if nativeID == "" { - // Fall back to drift entry if command response doesn't have it. - if _, drift, ok := model.managedDriftForResource(ru.StackName, ru.ResourceLabel); ok { - nativeID = drift.NativeID - } - } - if nativeID != "" { - if err := h.TryDeleteCloudState(nativeID); err != nil { - t.Logf("reconcileManagedDriftOverriddenByCommand: skipping DeleteCloudState for %s: %v", nativeID, err) - } - } - model.ClearManagedDriftForResource(ru.StackName, ru.ResourceLabel) - continue - } - // Create/Update: sync cloud state if this resource had a managed drift - // entry. The OOB drift operation may have deleted/modified cloud state, - // and the command restored the resource. The plugin's CRUD methods update - // cloud state with resolved properties, but for delete-drifted resources - // the NativeID may have changed. Use the model's NativeID tracking. - nativeID := ru.NativeID - if nativeID != "" { - if _, drift, ok := model.managedDriftForResource(ru.StackName, ru.ResourceLabel); ok { - // Restore cloud state: use command response properties flattened. - // This may contain partially resolved values but it's better than - // leaving the cloud state empty after an OOB delete. - if ru.Properties != nil { - if err := h.TryPutCloudState(nativeID, ru.ResourceType, flattenPropertiesForCloud(ru.Properties)); err != nil { - t.Logf("reconcileManagedDriftOverriddenByCommand: skipping PutCloudState for %s: %v", nativeID, err) - } - } - _ = drift // used for lookup only - } - } - model.ClearManagedDriftForResource(ru.StackName, ru.ResourceLabel) - } } func (h *TestHarness) reconcileAmbiguousFailedCommands(t *testing.T, model *StateModel, drained []drainedCommand) { diff --git a/tests/blackbox/harness.go b/tests/blackbox/harness.go index 5b0a4cb45..0942692a9 100644 --- a/tests/blackbox/harness.go +++ b/tests/blackbox/harness.go @@ -359,23 +359,27 @@ func (h *TestHarness) latestSyncCommand() (*apimodel.Command, error) { return &commands[0], nil } -// WaitForNextSyncCommand waits for a new sync command created after the current -// latest sync command, then waits for that command to reach a terminal state. -// Returns nil,false if no new sync command appears within the timeout. -func (h *TestHarness) WaitForNextSyncCommand(timeout time.Duration) (*apimodel.Command, bool) { +// SyncCommandBaseline returns the ID of the most recent sync command, or "" +// when none exists. Capture it BEFORE triggering a sync: the triggered sync +// command can be persisted before the first post-trigger poll runs, and a +// baseline taken after the trigger would then absorb the very command being +// waited for. +func (h *TestHarness) SyncCommandBaseline() string { before, err := h.latestSyncCommand() - if err != nil { - return nil, false - } - beforeID := "" - if before != nil { - beforeID = before.CommandID + if err != nil || before == nil { + return "" } + return before.CommandID +} +// WaitForSyncCommandAfter waits for a sync command different from the given +// baseline ID, then waits for that command to reach a terminal state. +// Returns nil,false if no such sync command appears within the timeout. +func (h *TestHarness) WaitForSyncCommandAfter(baselineID string, timeout time.Duration) (*apimodel.Command, bool) { deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { current, err := h.latestSyncCommand() - if err == nil && current != nil && current.CommandID != "" && current.CommandID != beforeID { + if err == nil && current != nil && current.CommandID != "" && current.CommandID != baselineID { return h.waitForCommandInDB(current.CommandID, time.Until(deadline)) } time.Sleep(50 * time.Millisecond) @@ -383,6 +387,17 @@ func (h *TestHarness) WaitForNextSyncCommand(timeout time.Duration) (*apimodel.C return nil, false } +// WaitForNextSyncCommand waits for a new sync command created after the current +// latest sync command, then waits for that command to reach a terminal state. +// Returns nil,false if no new sync command appears within the timeout. +// +// The baseline is taken at call time, so this only observes syncs triggered +// AFTER the call starts; when waiting on a sync you already triggered, use +// SyncCommandBaseline + WaitForSyncCommandAfter around the trigger instead. +func (h *TestHarness) WaitForNextSyncCommand(timeout time.Duration) (*apimodel.Command, bool) { + return h.WaitForSyncCommandAfter(h.SyncCommandBaseline(), timeout) +} + // --- TestController communication (via Ergo cross-node calls) --- // GetCloudStateSnapshot queries the test plugin's TestController for the @@ -614,7 +629,7 @@ func (h *TestHarness) KillAgent(t *testing.T) { // RestartAgent starts a new agent subprocess with the same config (same SQLite DB). // Reconstructs the plugin's cloud state from the agent's inventory and the OOB // mirror, re-programs response sequences, then opens the gate. -func (h *TestHarness) RestartAgent(t *testing.T, timeout time.Duration, model ...*StateModel) { +func (h *TestHarness) RestartAgent(t *testing.T, timeout time.Duration) { t.Helper() // Start the agent but DON'T open the gate yet — ReRunIncompleteCommands @@ -632,23 +647,12 @@ func (h *TestHarness) RestartAgent(t *testing.T, timeout time.Duration, model .. // resolvables before re-injection, otherwise the plugin returns $res objects // on Read, which corrupts the $ref.$value in mergeRefsPreservingUserRefs. forma, err := h.client.ExtractResources("managed:true") - pendingManagedProps := map[string]string{} - pendingManagedDeletes := map[string]bool{} - if len(model) > 0 && model[0] != nil { - pendingManagedProps, pendingManagedDeletes = model[0].PendingManagedDriftCloudState() - } if err == nil && forma != nil { for _, res := range forma.Resources { if res.NativeID == "" { continue } - if pendingManagedDeletes[res.NativeID] { - continue - } flatProps := flattenPropertiesForCloud(res.Properties) - if driftProps, ok := pendingManagedProps[res.NativeID]; ok { - flatProps = driftProps - } _, err := h.callTestController(testcontrol.PutCloudStateRequest{ NativeID: res.NativeID, ResourceType: res.Type, @@ -662,12 +666,6 @@ func (h *TestHarness) RestartAgent(t *testing.T, timeout time.Duration, model .. // Re-inject OOB cloud state from the mirror (resources not in inventory). for _, entry := range h.cloudStateMirror { - if _, ok := pendingManagedProps[entry.NativeID]; ok { - continue - } - if pendingManagedDeletes[entry.NativeID] { - continue - } _, err := h.callTestController(testcontrol.PutCloudStateRequest{ NativeID: entry.NativeID, ResourceType: entry.ResourceType, diff --git a/tests/blackbox/invariants.go b/tests/blackbox/invariants.go index a5e9fa4d4..67d6b7809 100644 --- a/tests/blackbox/invariants.go +++ b/tests/blackbox/invariants.go @@ -50,7 +50,7 @@ type CommandState struct { // 1. No phantom resources: every resource in inventory exists in cloud state // 2. No orphaned cloud resources: every cloud resource is tracked in inventory // 3. Property consistency: inventory properties match cloud state properties -func CheckInvariants(inventory []pkgmodel.Resource, cloudState map[string]testcontrol.CloudStateEntry, ignoreNativeIDs map[string]bool, ignoreManagedDriftNativeIDs map[string]bool) []Violation { +func CheckInvariants(inventory []pkgmodel.Resource, cloudState map[string]testcontrol.CloudStateEntry, ignoreNativeIDs map[string]bool) []Violation { var violations []Violation // Build lookup of inventory resources by NativeID. @@ -87,9 +87,6 @@ func CheckInvariants(inventory []pkgmodel.Resource, cloudState map[string]testco if res.NativeID == "" { continue } - if ignoreManagedDriftNativeIDs[res.NativeID] { - continue - } if _, ok := cloudState[res.NativeID]; !ok { violations = append(violations, Violation{ Kind: ViolationPhantomResource, @@ -116,9 +113,6 @@ func CheckInvariants(inventory []pkgmodel.Resource, cloudState map[string]testco // Invariant 3: Property consistency // For resources present in both, properties should match for nativeID, cloudEntry := range cloudState { - if ignoreManagedDriftNativeIDs[nativeID] { - continue - } invRes, ok := inventoryByNativeID[nativeID] if !ok { continue // already reported as orphaned or ignored @@ -518,9 +512,6 @@ func CheckModelVsInventory(model *StateModel, inventory []pkgmodel.Resource) []V } label := model.LabelForResource(s, idx) resourceType := model.TypeForResource(idx) - if model.HasPendingManagedDriftAffectingSlot(s, idx) { - continue - } key := stack.Label + "/" + label invRes, existsInInventory := inventoryByKey[key] @@ -581,10 +572,7 @@ func CheckModelVsInventory(model *StateModel, inventory []pkgmodel.Resource) []V if expectedExistingKeys[key] { continue } - stackIdx, slotIdx, ok := model.findResourceSlot(res.Stack, res.Label) - if ok && model.HasPendingManagedDriftAffectingSlot(stackIdx, slotIdx) { - continue - } + _, slotIdx, ok := model.findResourceSlot(res.Stack, res.Label) if ok && model.Pool != nil && model.Pool.IsCrossStack(slotIdx) { continue } @@ -715,65 +703,6 @@ func CheckUnmanagedModelVsInventory(model *StateModel, inventory []pkgmodel.Reso return violations } -func CheckManagedDriftVsInventory(model *StateModel, inventory []pkgmodel.Resource) []Violation { - var violations []Violation - actualByNativeID := make(map[string]pkgmodel.Resource, len(inventory)) - for _, res := range inventory { - if res.NativeID == "" { - continue - } - actualByNativeID[res.NativeID] = res - } - - for nativeID, expected := range model.ManagedDriftedResources { - if expected.PendingSync { - continue - } - actual, ok := actualByNativeID[nativeID] - if !ok { - if expected.PresentInInventory { - violations = append(violations, Violation{ - Kind: ViolationModelInventoryMismatch, - Message: fmt.Sprintf("managed drift resource %s expected in inventory but missing", nativeID), - }) - } - continue - } - if !actual.Managed { - violations = append(violations, Violation{ - Kind: ViolationModelInventoryMismatch, - Message: fmt.Sprintf("managed drift resource %s unexpectedly marked unmanaged in inventory", nativeID), - }) - } - if expected.StackLabel != "" && actual.Stack != expected.StackLabel { - violations = append(violations, Violation{ - Kind: ViolationModelInventoryMismatch, - Message: fmt.Sprintf("managed drift resource %s stack=%s, expected %s", nativeID, actual.Stack, expected.StackLabel), - }) - } - if expected.ResourceLabel != "" && actual.Label != expected.ResourceLabel { - violations = append(violations, Violation{ - Kind: ViolationModelInventoryMismatch, - Message: fmt.Sprintf("managed drift resource %s label=%s, expected %s", nativeID, actual.Label, expected.ResourceLabel), - }) - } - if expected.ResourceType != "" && actual.Type != expected.ResourceType { - violations = append(violations, Violation{ - Kind: ViolationModelInventoryMismatch, - Message: fmt.Sprintf("managed drift resource %s type=%s, expected %s", nativeID, actual.Type, expected.ResourceType), - }) - } - if expected.InventoryProperties != "" && !jsonEqual(string(actual.Properties), expected.InventoryProperties) { - violations = append(violations, Violation{ - Kind: ViolationModelInventoryMismatch, - Message: fmt.Sprintf("managed drift resource %s inventory properties=%s but model expects %s", nativeID, string(actual.Properties), expected.InventoryProperties), - }) - } - } - - return violations -} - // CheckOperationLogInvariants validates stable plugin-side facts. These are // intentionally conservative so they hold across crash recovery and restart // paths where end-state existence cannot be inferred from the operation log diff --git a/tests/blackbox/invariants_rename_test.go b/tests/blackbox/invariants_rename_test.go index 5824dec97..61137aca3 100644 --- a/tests/blackbox/invariants_rename_test.go +++ b/tests/blackbox/invariants_rename_test.go @@ -29,7 +29,7 @@ func TestCheckInvariants_FlagsDuplicateNativeID(t *testing.T) { {Ksuid: "K_B", NativeID: "i-0abc1234", Label: "new-vpc", Type: "AWS::EC2::Instance"}, } - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) var foundDup bool for _, v := range violations { @@ -53,7 +53,7 @@ func TestCheckInvariants_SingleRowPerNativeID_OK(t *testing.T) { {Ksuid: "K_A", NativeID: "i-0abc1234", Label: "app-server", Type: "AWS::EC2::Instance"}, } - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) for _, v := range violations { if v.Kind == ViolationDuplicateNativeID { diff --git a/tests/blackbox/property_test.go b/tests/blackbox/property_test.go index db0d0cf01..1fac028f8 100644 --- a/tests/blackbox/property_test.go +++ b/tests/blackbox/property_test.go @@ -171,7 +171,7 @@ func TestProperty_FullChaos(t *testing.T) { // Sync cloud state with inventory. Chaos operations (cancel, crash, // TTL cascade) can leave cloud entries that the ResourcePersister // hasn't cleaned up yet, causing phantom violations. - h.TriggerSyncAndWait(t) + h.TriggerSyncAndWait(t, model) h.AssertAllInvariants(t, model) }) diff --git a/tests/blackbox/state_model.go b/tests/blackbox/state_model.go index 2b1c1ab99..99db09c9f 100644 --- a/tests/blackbox/state_model.go +++ b/tests/blackbox/state_model.go @@ -11,9 +11,6 @@ import ( "fmt" "sort" "strings" - - apimodel "github.com/platform-engineering-labs/formae/pkg/api/model" - pkgmodel "github.com/platform-engineering-labs/formae/pkg/model" ) // ResourceState represents whether a resource is expected to exist. @@ -42,22 +39,6 @@ type ExpectedUnmanagedResource struct { PresentInInventory bool } -// ExpectedManagedDrift tracks the cloud-side state of a managed resource after -// out-of-band changes and the inventory state observed after sync. -type ExpectedManagedDrift struct { - StackLabel string - ResourceLabel string - ResourceType string - NativeID string - CloudProperties string - InventoryProperties string - SnapshotProperties string - SnapshotState ResourceState - PresentInCloud bool - PresentInInventory bool - PendingSync bool -} - // StackState holds the per-stack state: resources. type StackState struct { Label string @@ -86,9 +67,6 @@ type StateModel struct { // UnmanagedResources tracks out-of-band cloud resources and whether they are // expected to have been ingested into the unmanaged inventory. UnmanagedResources map[string]*ExpectedUnmanagedResource - // ManagedDriftedResources tracks managed resources that have been mutated or - // deleted out-of-band so sync behavior can be asserted explicitly. - ManagedDriftedResources map[string]*ExpectedManagedDrift AuthoritativeSlots map[string]bool // NativeIDs maps "stackIdx:slotIdx" → cloud native ID (e.g. "test-42"). // Populated from command response ResourceUpdate.NativeID on successful @@ -142,14 +120,13 @@ func NewStateModel(stackCount, resourcesPerStack int) *StateModel { providerLabel = stacks[0].Label } return &StateModel{ - Stacks: stacks, - ResourcesPerStack: resourcesPerStack, - Pool: pool, - ProviderStackLabel: providerLabel, - UnmanagedResources: make(map[string]*ExpectedUnmanagedResource), - ManagedDriftedResources: make(map[string]*ExpectedManagedDrift), - AuthoritativeSlots: make(map[string]bool), - NativeIDs: make(map[string]string), + Stacks: stacks, + ResourcesPerStack: resourcesPerStack, + Pool: pool, + ProviderStackLabel: providerLabel, + UnmanagedResources: make(map[string]*ExpectedUnmanagedResource), + AuthoritativeSlots: make(map[string]bool), + NativeIDs: make(map[string]string), } } @@ -175,18 +152,43 @@ func (m *StateModel) ClearNativeID(stackIdx, slotIdx int) { delete(m.NativeIDs, nativeIDKey(stackIdx, slotIdx)) } -// FindExistingResourceWithNativeID finds a managed resource that exists in the -// model and has a tracked NativeID. Used for selecting OOB drift targets. -// The sequenceNum is used for deterministic selection (modulo eligible count). -func (m *StateModel) FindExistingResourceWithNativeID(sequenceNum int) (stackIdx, slotIdx int, stackLabel, resourceLabel, resourceType, nativeID string, ok bool) { +// FindDriftEligibleResource finds a managed resource that exists in the model, +// has a tracked NativeID, and is untouched by in-flight commands. Used for +// selecting OOB drift targets: drift is absorbed synchronously via a forced +// sync, which only converges deterministically for resources no in-flight +// command can create, update, or (via the reconcile guarantee) delete. +// +// A slot is excluded when its stack has an accepted command, and a cross-stack +// slot is additionally excluded while the provider stack (stack 0) has one, +// because provider-stack cascades reach cross-stack dependents. +// +// The sequenceNum selects deterministically (modulo eligible count) from a +// stably ordered candidate list. +func (m *StateModel) FindDriftEligibleResource(sequenceNum int) (stackIdx, slotIdx int, stackLabel, resourceLabel, resourceType, nativeID string, ok bool) { + busyStacks := make(map[int]bool) + for _, ac := range m.AcceptedCommands { + for _, ref := range ac.RequestedSlots { + busyStacks[ref.StackIndex] = true + } + } + type candidate struct { - stackIdx, slotIdx int - stackLabel, label, rType string - nativeID string + stackIdx, slotIdx int + stackLabel, label, rType string + nativeID string } var eligible []candidate for si, stack := range m.Stacks { - for idx, res := range stack.Resources { + if busyStacks[si] { + continue + } + slots := make([]int, 0, len(stack.Resources)) + for idx := range stack.Resources { + slots = append(slots, idx) + } + sort.Ints(slots) + for _, idx := range slots { + res := stack.Resources[idx] if res == nil || res.State != StateExists { continue } @@ -194,6 +196,9 @@ func (m *StateModel) FindExistingResourceWithNativeID(sequenceNum int) (stackIdx if nid == "" { continue } + if m.Pool != nil && m.Pool.IsCrossStack(idx) && busyStacks[0] { + continue + } var label string if m.Pool != nil { label = m.Pool.LabelForStack(stack.Label, idx) @@ -260,208 +265,20 @@ func (m *StateModel) ClearAuthorityForStack(stackIdx int) { } } -func (m *StateModel) ApplyManagedCloudModify(stackLabel, resourceLabel, resourceType, nativeID, properties string) { - if nativeID == "" { - return - } - res := m.ManagedDriftedResources[nativeID] - if res == nil { - res = &ExpectedManagedDrift{NativeID: nativeID} - if stackIdx, slotIdx, ok := m.findResourceSlot(stackLabel, resourceLabel); ok { - if existing := m.Resource(stackIdx, slotIdx); existing != nil { - res.SnapshotProperties = existing.Properties - res.SnapshotState = existing.State - } - } - m.ManagedDriftedResources[nativeID] = res - } - res.StackLabel = stackLabel - res.ResourceLabel = resourceLabel - res.ResourceType = resourceType - res.CloudProperties = properties - res.PresentInCloud = true - res.PendingSync = true -} - -func (m *StateModel) ApplyManagedCloudDelete(stackLabel, resourceLabel, resourceType, nativeID string) { - if nativeID == "" { - return - } - res := m.ManagedDriftedResources[nativeID] - if res == nil { - res = &ExpectedManagedDrift{NativeID: nativeID} - if stackIdx, slotIdx, ok := m.findResourceSlot(stackLabel, resourceLabel); ok { - if existing := m.Resource(stackIdx, slotIdx); existing != nil { - res.SnapshotProperties = existing.Properties - res.SnapshotState = existing.State - } - } - m.ManagedDriftedResources[nativeID] = res - } - res.StackLabel = stackLabel - res.ResourceLabel = resourceLabel - res.ResourceType = resourceType - res.CloudProperties = "" - res.PresentInCloud = false - res.PendingSync = true -} - -func (m *StateModel) ApplySyncToManagedDrift() { - for nativeID, expected := range m.ManagedDriftedResources { - expected.PendingSync = false - if expected.PresentInCloud { - actual := pkgmodel.Resource{ - Stack: expected.StackLabel, - Label: expected.ResourceLabel, - Type: expected.ResourceType, - NativeID: expected.NativeID, - Managed: true, - Properties: []byte(expected.CloudProperties), - } - m.applyManagedDriftToResource(expected, &actual) - } else { - m.applyManagedDriftToResource(expected, nil) - } - delete(m.ManagedDriftedResources, nativeID) - } -} - -func (m *StateModel) ApplySyncCommand(cmd *apimodel.Command) { - if cmd == nil { - return - } - for _, ru := range cmd.ResourceUpdates { - if ru.State != "Success" { - continue - } - if driftNativeID, drift, ok := m.managedDriftForResource(ru.StackName, ru.ResourceLabel); ok { - if ru.Operation == "delete" { - drift.PresentInCloud = false - m.applyManagedDriftToResource(drift, nil) - } else if ru.Properties != nil { - drift.PresentInCloud = true - drift.CloudProperties = string(ru.Properties) - actual := pkgmodel.Resource{Stack: ru.StackName, Label: ru.ResourceLabel, Type: ru.ResourceType, Managed: true, Properties: ru.Properties} - m.applyManagedDriftToResource(drift, &actual) - } - drift.PendingSync = false - delete(m.ManagedDriftedResources, driftNativeID) - } - } -} - -func (m *StateModel) ApplyDiscoveryCommand(cmd *apimodel.Command) { - m.ApplyDiscoveryToUnmanaged() -} - -func (m *StateModel) managedDriftForResource(stackLabel, resourceLabel string) (string, *ExpectedManagedDrift, bool) { - for nativeID, drift := range m.ManagedDriftedResources { - if drift.StackLabel == stackLabel && drift.ResourceLabel == resourceLabel { - return nativeID, drift, true - } - } - return "", nil, false -} - -func (m *StateModel) PendingManagedDriftNativeIDs() map[string]bool { - ignore := make(map[string]bool) - for nativeID, res := range m.ManagedDriftedResources { - if res.PendingSync { - ignore[nativeID] = true - } - } - return ignore -} - -func (m *StateModel) ManagedDriftNativeIDs() map[string]bool { - ignore := make(map[string]bool) - for nativeID := range m.ManagedDriftedResources { - ignore[nativeID] = true - } - return ignore -} - -func (m *StateModel) PendingManagedDriftCloudState() (map[string]string, map[string]bool) { - propsByNativeID := make(map[string]string) - deleted := make(map[string]bool) - for nativeID, res := range m.ManagedDriftedResources { - if !res.PendingSync { - continue - } - if res.PresentInCloud { - propsByNativeID[nativeID] = res.CloudProperties - } else { - deleted[nativeID] = true - } - } - return propsByNativeID, deleted -} - -func (m *StateModel) HasPendingManagedDriftForResource(stackLabel, resourceLabel string) bool { - for _, res := range m.ManagedDriftedResources { - if res.PendingSync && res.StackLabel == stackLabel && res.ResourceLabel == resourceLabel { - return true - } - } - return false -} - -func (m *StateModel) HasPendingManagedDriftAffectingSlot(stackIdx, slotIdx int) bool { - stack := m.Stack(stackIdx) - if stack == nil { - return false - } - label := m.LabelForResource(stackIdx, slotIdx) - if m.HasPendingManagedDriftForResource(stack.Label, label) { - return true - } - if m.Pool == nil { - return false - } - current := slotIdx - for current >= 0 { - slot := m.Pool.Slots[current] - if slot.ParentIndex >= 0 { - parentLabel := m.LabelForResource(stackIdx, slot.ParentIndex) - if m.HasPendingManagedDriftForResource(stack.Label, parentLabel) { - return true - } - } - current = slot.ParentIndex - } - if m.Pool.IsCrossStack(slotIdx) && m.ProviderStackLabel != "" { - parentSlot := m.Pool.Slots[slotIdx].CrossStackParentSlot - if parentSlot >= 0 && m.HasPendingManagedDriftForResource(m.ProviderStackLabel, m.LabelForResource(0, parentSlot)) { - return true - } - } - return false -} - -func (m *StateModel) ClearManagedDriftForResource(stackLabel, resourceLabel string) { - for nativeID, res := range m.ManagedDriftedResources { - if res.StackLabel == stackLabel && res.ResourceLabel == resourceLabel { - delete(m.ManagedDriftedResources, nativeID) - } - } +// ApplyManagedCloudModify records the deterministic end state of an +// out-of-band modify on a managed resource: the agent's sync absorbs the +// drift, so inventory converges on the cloud properties. +func (m *StateModel) ApplyManagedCloudModify(stackIdx, slotIdx int, properties string) { + props := m.NormalizePropertiesForResource(stackIdx, slotIdx, properties) + m.ApplyCreated(stackIdx, []int{slotIdx}, props) } -func (m *StateModel) applyManagedDriftToResource(expected *ExpectedManagedDrift, actual *pkgmodel.Resource) { - stackIdx, slotIdx, ok := m.findResourceSlot(expected.StackLabel, expected.ResourceLabel) - if !ok { - return - } - - if actual == nil { - m.ApplyDestroyed(stackIdx, []int{slotIdx}) - m.MarkAuthoritativeSlot(stackIdx, slotIdx) - m.refreshDependentParentReferences(stackIdx, slotIdx) - return - } - props := m.NormalizePropertiesForResource(stackIdx, slotIdx, string(actual.Properties)) - m.ApplyCreated(stackIdx, []int{slotIdx}, props) - m.MarkAuthoritativeSlot(stackIdx, slotIdx) - m.refreshDependentParentReferences(stackIdx, slotIdx) +// ApplyManagedCloudDelete records the deterministic end state of an +// out-of-band delete of a managed resource: the agent's sync observes the +// deletion and drops the resource from inventory. +func (m *StateModel) ApplyManagedCloudDelete(stackIdx, slotIdx int) { + m.ApplyDestroyed(stackIdx, []int{slotIdx}) + m.ClearNativeID(stackIdx, slotIdx) } func (m *StateModel) findResourceSlot(stackLabel, resourceLabel string) (int, int, bool) { @@ -483,49 +300,6 @@ func (m *StateModel) findResourceSlot(stackLabel, resourceLabel string) (int, in return -1, -1, false } -func (m *StateModel) refreshDependentParentReferences(stackIdx, slotIdx int) { - if m.Pool == nil { - return - } - for dependent := range m.Stack(stackIdx).Resources { - if dependent == slotIdx { - continue - } - if m.Pool.Slots[dependent].ParentIndex != slotIdx { - continue - } - res := m.Resource(stackIdx, dependent) - if res == nil || res.State != StateExists || res.Properties == "" { - continue - } - updated := m.NormalizePropertiesForResource(stackIdx, dependent, res.Properties) - res.Properties = updated - m.refreshDependentParentReferences(stackIdx, dependent) - } - - if m.ProviderStackLabel == "" || m.Stack(stackIdx).Label != m.ProviderStackLabel { - return - } - for consumerStackIdx := range m.Stacks { - if consumerStackIdx == stackIdx { - continue - } - for dependent := range m.Stack(consumerStackIdx).Resources { - if !m.Pool.IsCrossStack(dependent) { - continue - } - if m.Pool.Slots[dependent].CrossStackParentSlot != slotIdx { - continue - } - res := m.Resource(consumerStackIdx, dependent) - if res == nil || res.State != StateExists || res.Properties == "" { - continue - } - res.Properties = m.NormalizePropertiesForResource(consumerStackIdx, dependent, res.Properties) - } - } -} - func (m *StateModel) EnsureUnmanagedResource(nativeID, resourceType, properties string) *ExpectedUnmanagedResource { if existing, ok := m.UnmanagedResources[nativeID]; ok { if resourceType != "" { diff --git a/tests/blackbox/state_model_test.go b/tests/blackbox/state_model_test.go index 741eebbcc..d72098c66 100644 --- a/tests/blackbox/state_model_test.go +++ b/tests/blackbox/state_model_test.go @@ -7,6 +7,7 @@ package blackbox import ( + "fmt" "testing" "github.com/stretchr/testify/assert" @@ -150,7 +151,7 @@ func TestStateModel_Verify_HappyPath(t *testing.T) { "native-1": {NativeID: "native-1", ResourceType: "Test::Generic::Resource", Properties: `{"Name":"a","Value":"v1"}`}, } - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) assert.Empty(t, violations) } @@ -161,7 +162,7 @@ func TestStateModel_Verify_PhantomResource(t *testing.T) { } cloudState := map[string]testcontrol.CloudStateEntry{} - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) assert.NotEmpty(t, violations) hasPhantom := false @@ -180,7 +181,7 @@ func TestStateModel_Verify_OrphanedResource(t *testing.T) { "native-0": {NativeID: "native-0", ResourceType: "Test::Generic::Resource"}, } - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) assert.NotEmpty(t, violations) hasOrphan := false @@ -202,7 +203,7 @@ func TestStateModel_Verify_PropertyMismatch(t *testing.T) { Properties: `{"Name":"a","Value":"v2"}`}, } - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) assert.NotEmpty(t, violations) hasMismatch := false @@ -224,7 +225,7 @@ func TestStateModel_Verify_PropertyMatch(t *testing.T) { Properties: `{"Name":"a","Value":"v1"}`}, } - violations := CheckInvariants(inventory, cloudState, nil, nil) + violations := CheckInvariants(inventory, cloudState, nil) assert.Empty(t, violations) } @@ -389,23 +390,120 @@ func TestStateModel_UnmanagedLifecycle(t *testing.T) { assert.False(t, res.PresentInCloud) } -func TestStateModel_ManagedDriftLifecycle(t *testing.T) { +// An out-of-band modify of a managed resource is absorbed by sync: inventory +// converges on the cloud properties. The model computes that end state +// directly at the drift operation. +func TestStateModel_ManagedCloudModifyComputesAbsorbedState(t *testing.T) { model := NewStateModel(1, 1) model.ApplyCreated(0, []int{0}, `{"Name":"res-stack-0-a","Value":"v1"}`) - model.ApplyManagedCloudModify("stack-0", "res-stack-0-a", "Test::Generic::Resource", "test-1", `{"Name":"res-stack-0-a","Value":"drift"}`) + model.SetNativeID(0, 0, "test-1") - // Sync picks up the drift and applies it to the model. - model.ApplySyncToManagedDrift() + model.ApplyManagedCloudModify(0, 0, `{"Name":"cloud-res-7","Value":"drift"}`) res := model.Resource(0, 0) assert.Equal(t, StateExists, res.State) - assert.JSONEq(t, `{"Name":"res-stack-0-a","Value":"drift"}`, res.Properties) - assert.Empty(t, model.ManagedDriftedResources) + assert.JSONEq(t, `{"Name":"cloud-res-7","Value":"drift"}`, res.Properties) + assert.Equal(t, "test-1", model.GetNativeID(0, 0), "modify keeps the native id") +} - model.ApplyManagedCloudDelete("stack-0", "res-stack-0-a", "Test::Generic::Resource", "test-1") - model.ApplySyncToManagedDrift() - assert.Equal(t, StateNotExist, model.Resource(0, 0).State) - assert.Empty(t, model.ManagedDriftedResources) +// An out-of-band delete of a managed resource is absorbed by sync: the agent +// drops the resource from inventory. The model computes that end state +// directly at the drift operation. +func TestStateModel_ManagedCloudDeleteComputesAbsorbedState(t *testing.T) { + model := NewStateModel(1, 1) + model.ApplyCreated(0, []int{0}, `{"Name":"res-stack-0-a","Value":"v1"}`) + model.SetNativeID(0, 0, "test-1") + + model.ApplyManagedCloudDelete(0, 0) + + res := model.Resource(0, 0) + assert.Equal(t, StateNotExist, res.State) + assert.Empty(t, res.Properties) + assert.Empty(t, model.GetNativeID(0, 0), "delete clears the native id") +} + +// Drift targets must not be touched by in-flight commands: a command on a +// stack can create, update, or (via the reconcile guarantee) delete any slot +// on that stack, so drift on such a slot would race the command. +func TestStateModel_DriftEligibilitySkipsStacksWithInFlightCommands(t *testing.T) { + model := NewStateModel(2, 3) + model.ApplyCreated(0, []int{0}, `{"Name":"res-stack-0-a","Value":"v1"}`) + model.SetNativeID(0, 0, "test-1") + model.ApplyCreated(1, []int{0}, `{"Name":"res-stack-1-a","Value":"v1"}`) + model.SetNativeID(1, 0, "test-2") + + model.TrackAcceptedCommand("cmd-1", nil, []ResourceSlotRef{{StackIndex: 1, SlotIndex: 0}}, 0, false) + + for seq := range 4 { + stackIdx, _, _, _, _, nativeID, ok := model.FindDriftEligibleResource(seq) + require.True(t, ok) + assert.Equal(t, 0, stackIdx, "stack 1 has an in-flight command and must be excluded") + assert.Equal(t, "test-1", nativeID) + } + + model.AcceptedCommands = nil + seen := make(map[int]bool) + for seq := range 4 { + stackIdx, _, _, _, _, _, ok := model.FindDriftEligibleResource(seq) + require.True(t, ok) + seen[stackIdx] = true + } + assert.True(t, seen[0] && seen[1], "with no in-flight commands both stacks are eligible") +} + +// Cross-stack slots reference a parent on the provider stack (stack 0), so an +// in-flight command on the provider stack can cascade onto them. They are only +// eligible for drift while the provider stack is quiescent too. +func TestStateModel_DriftEligibilityCrossStackNeedsQuiescentProvider(t *testing.T) { + model := NewStateModel(2, 10) + require.NotNil(t, model.Pool) + + crossIdx := -1 + for i := range model.Pool.Slots { + if model.Pool.IsCrossStack(i) { + crossIdx = i + break + } + } + require.GreaterOrEqual(t, crossIdx, 0) + + model.ApplyCreated(1, []int{crossIdx}, `{"Name":"x","ParentId":"p","Value":"v1"}`) + model.SetNativeID(1, crossIdx, "test-9") + + // A command on the provider stack (slot 0 there) excludes cross-stack slots. + model.TrackAcceptedCommand("cmd-1", nil, []ResourceSlotRef{{StackIndex: 0, SlotIndex: 0}}, 0, false) + _, _, _, _, _, _, ok := model.FindDriftEligibleResource(0) + assert.False(t, ok, "the only existing resource is cross-stack and the provider stack is busy") + + model.AcceptedCommands = nil + stackIdx, slotIdx, _, _, _, nativeID, ok := model.FindDriftEligibleResource(0) + require.True(t, ok) + assert.Equal(t, 1, stackIdx) + assert.Equal(t, crossIdx, slotIdx) + assert.Equal(t, "test-9", nativeID) +} + +// Eligible drift targets must be enumerated in a stable order: selection is +// keyed off the generated sequence number, and map-iteration order would make +// the picked resource nondeterministic across runs. +func TestStateModel_DriftEligibilityOrderIsDeterministic(t *testing.T) { + model := NewStateModel(2, 3) + for s := range 2 { + for i := range 3 { + model.ApplyCreated(s, []int{i}, `{"Name":"n","Value":"v1"}`) + model.SetNativeID(s, i, fmt.Sprintf("test-%d%d", s, i)) + } + } + + for seq := range 6 { + _, _, _, _, _, first, ok := model.FindDriftEligibleResource(seq) + require.True(t, ok) + for range 20 { + _, _, _, _, _, again, ok := model.FindDriftEligibleResource(seq) + require.True(t, ok) + require.Equal(t, first, again) + } + } } func TestStateModel_Stack(t *testing.T) { From 2b850fc766a406ec1189922d79ca2311d891af83 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Mon, 17 Aug 2026 15:52:47 -0700 Subject: [PATCH 2/3] test(blackbox): observe syncs through their deletion window and stop waiting on impossible commands A sync command whose reads produce no changes is deleted from the datastore on completion, so a no-op sync is unobservable by design. The harness's awaited syncs burned their full timeout on every no-op sync, and the discovery trigger waited ten seconds for an ingestion command that can never exist: the suite's formas never mark the target Discoverable, so discovery finds no discoverable targets and ingests nothing. - split the sync wait into a short appearance window (command creation is milliseconds) and a separate completion window, and treat a command that appears and then vanishes as an observed no-change sync - make the discovery trigger fire-and-forget with no model transition: with no discoverable targets the deterministic end state is no change, and any surprise ingestion still surfaces as an unexpected unmanaged inventory row - previously the discovery trigger applied the ingestion transition whenever any sync command happened to be observed within its window, setting expectations for rows that were never ingested --- tests/blackbox/executor.go | 49 +++++++++++++++++------------------ tests/blackbox/harness.go | 52 ++++++++++++++++++++++++++++++++------ 2 files changed, 69 insertions(+), 32 deletions(-) diff --git a/tests/blackbox/executor.go b/tests/blackbox/executor.go index 66825d570..ae3ba8b56 100644 --- a/tests/blackbox/executor.go +++ b/tests/blackbox/executor.go @@ -1569,7 +1569,7 @@ func (h *TestHarness) executeTriggerSync(t *testing.T, model *StateModel) { // // An unobserved sync is tolerable here: with no drift pending by // construction, the sync's model transition is a no-op either way. - if h.forceSyncAndAwait(t, model) { + if h.forceSyncAndAwait(t, model, 2*time.Second) { t.Logf("TriggerSync: fired") } else { t.Logf("TriggerSync: no sync command observed (nothing to synchronize)") @@ -1582,25 +1582,28 @@ func (h *TestHarness) executeTriggerSync(t *testing.T, model *StateModel) { // As in executeTriggerSync, an unobserved sync is a model no-op. func (h *TestHarness) TriggerSyncAndWait(t *testing.T, model *StateModel) { t.Helper() - if !h.forceSyncAndAwait(t, model) { + if !h.forceSyncAndAwait(t, model, 2*time.Second) { t.Logf("TriggerSyncAndWait: no sync command observed (nothing to synchronize)") } } -// forceSyncAndAwait triggers a sync and waits for its command to reach a -// terminal state. When observed, it applies the model's own predicted sync -// transition for unmanaged rows (sync reads every idle inventory row, so -// out-of-band changes to discovered unmanaged resources absorb at each sync) -// and returns true. It returns false when no sync command appeared: the -// synchronizer creates none when it finds nothing to synchronize (empty -// inventory, or every row filtered by an in-flight exclusion or a transient -// plugin-availability miss). The baseline is captured before the trigger so -// the triggered command cannot be mistaken for a pre-existing one. -func (h *TestHarness) forceSyncAndAwait(t *testing.T, model *StateModel) bool { +// forceSyncAndAwait triggers a sync and waits for its command to complete. +// When observed, it applies the model's own predicted sync transition for +// unmanaged rows (sync reads every idle inventory row, so out-of-band changes +// to discovered unmanaged resources absorb at each sync) and returns true. +// +// It returns false when no sync command was observed within the appearance +// window. That covers both a sync that never ran (empty inventory, transient +// plugin-availability miss, synchronizer busy) and a no-change sync that +// completed and was deleted between polls — the two are indistinguishable, so +// callers may only treat false as tolerable where a no-op sync and no sync +// are equivalent. The baseline is captured before the trigger so the +// triggered command cannot be mistaken for a pre-existing one. +func (h *TestHarness) forceSyncAndAwait(t *testing.T, model *StateModel, appearance time.Duration) bool { t.Helper() baseline := h.SyncCommandBaseline() require.NoError(t, h.client.ForceSync(), "ForceSync failed") - _, ok := h.WaitForSyncCommandAfter(baseline, 15*time.Second) + _, ok := h.WaitForSyncCommandAfter(baseline, appearance, 30*time.Second) if !ok { return false } @@ -1612,18 +1615,16 @@ func (h *TestHarness) forceSyncAndAwait(t *testing.T, model *StateModel) bool { func (h *TestHarness) executeTriggerDiscovery(t *testing.T, model *StateModel) { t.Helper() - err := h.client.ForceDiscover() - if err != nil { + // The suite's formas never mark the target Discoverable, so discovery + // finds no discoverable targets and ingests nothing: the deterministic + // model transition is no transition at all. Fire it to exercise the + // path; any surprise ingestion surfaces as an unexpected unmanaged row + // in CheckUnmanagedModelVsInventory. + if err := h.client.ForceDiscover(); err != nil { t.Logf("TriggerDiscovery error (may be expected): %v", err) return } - if _, ok := h.WaitForNextSyncCommand(10 * time.Second); !ok { - t.Logf("TriggerDiscovery: no new discovery/sync command observed") - return - } else if model != nil { - model.ApplyDiscoveryToUnmanaged() - } - t.Logf("TriggerDiscovery: fired") + t.Logf("TriggerDiscovery: fired (no discoverable targets, ingests nothing)") } func (h *TestHarness) executeCloudModify(t *testing.T, op *Operation, model *StateModel) { @@ -1742,7 +1743,7 @@ func (h *TestHarness) absorbManagedDrift(t *testing.T, model *StateModel, native t.Helper() const maxSyncAttempts = 3 for attempt := range maxSyncAttempts { - if !h.forceSyncAndAwait(t, model) { + if !h.forceSyncAndAwait(t, model, 10*time.Second) { // The drifted row is in inventory, so a healthy sync must include // it; an unobserved sync command here is a transient miss — retry. t.Logf("absorbManagedDrift: no sync command observed (attempt %d)", attempt+1) @@ -1763,7 +1764,7 @@ func (h *TestHarness) absorbUnmanagedDrift(t *testing.T, model *StateModel, nati t.Helper() const maxSyncAttempts = 3 for attempt := range maxSyncAttempts { - if !h.forceSyncAndAwait(t, model) { + if !h.forceSyncAndAwait(t, model, 10*time.Second) { t.Logf("absorbUnmanagedDrift: no sync command observed (attempt %d)", attempt+1) continue } diff --git a/tests/blackbox/harness.go b/tests/blackbox/harness.go index 0942692a9..428cc9f06 100644 --- a/tests/blackbox/harness.go +++ b/tests/blackbox/harness.go @@ -372,21 +372,57 @@ func (h *TestHarness) SyncCommandBaseline() string { return before.CommandID } -// WaitForSyncCommandAfter waits for a sync command different from the given -// baseline ID, then waits for that command to reach a terminal state. -// Returns nil,false if no such sync command appears within the timeout. -func (h *TestHarness) WaitForSyncCommandAfter(baselineID string, timeout time.Duration) (*apimodel.Command, bool) { - deadline := time.Now().Add(timeout) - for time.Now().Before(deadline) { +// WaitForSyncCommandAfter waits up to `appearance` for a sync command +// different from the given baseline ID to be persisted, then up to +// `completion` for that command to finish. Returns nil,false if no new sync +// command appears within the appearance window. +// +// A sync command whose reads produce no changes is DELETED from the datastore +// on completion (the persister keeps only syncs that absorbed something), so +// a command that appears and then vanishes has completed as a no-change sync: +// that is reported as observed with a nil command. A no-change sync can also +// complete and be deleted entirely between polls, in which case it is +// indistinguishable from no sync at all — callers must only rely on a false +// return where a no-op sync and no sync are equivalent. +func (h *TestHarness) WaitForSyncCommandAfter(baselineID string, appearance, completion time.Duration) (*apimodel.Command, bool) { + appearDeadline := time.Now().Add(appearance) + for time.Now().Before(appearDeadline) { current, err := h.latestSyncCommand() if err == nil && current != nil && current.CommandID != "" && current.CommandID != baselineID { - return h.waitForCommandInDB(current.CommandID, time.Until(deadline)) + return h.waitForSyncCommandDone(current.CommandID, completion) } time.Sleep(50 * time.Millisecond) } return nil, false } +// waitForSyncCommandDone polls a known sync command until it reaches a +// terminal state or is deleted as a no-change sync (both count as observed). +func (h *TestHarness) waitForSyncCommandDone(commandID string, timeout time.Duration) (*apimodel.Command, bool) { + deadline := time.Now().Add(timeout) + seen := false + for time.Now().Before(deadline) { + cmd, err := h.commandFromDB(commandID) + if err == nil { + if cmd == nil { + if seen { + return nil, true // deleted on completion: a no-change sync + } + } else { + seen = true + h.ObserveCommandState(h.t, cmd.CommandID, cmd.State) + if isTerminalCommandState(cmd.State) { + return cmd, true + } + } + } + time.Sleep(50 * time.Millisecond) + } + // The row was observed at least once (latestSyncCommand returned it), so + // its disappearance without a terminal read still means it completed. + return nil, !seen +} + // WaitForNextSyncCommand waits for a new sync command created after the current // latest sync command, then waits for that command to reach a terminal state. // Returns nil,false if no new sync command appears within the timeout. @@ -395,7 +431,7 @@ func (h *TestHarness) WaitForSyncCommandAfter(baselineID string, timeout time.Du // AFTER the call starts; when waiting on a sync you already triggered, use // SyncCommandBaseline + WaitForSyncCommandAfter around the trigger instead. func (h *TestHarness) WaitForNextSyncCommand(timeout time.Duration) (*apimodel.Command, bool) { - return h.WaitForSyncCommandAfter(h.SyncCommandBaseline(), timeout) + return h.WaitForSyncCommandAfter(h.SyncCommandBaseline(), timeout, timeout) } // --- TestController communication (via Ergo cross-node calls) --- From d44ba7c9e84a7dd4e213339069c7047f30966a22 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Mon, 17 Aug 2026 16:04:16 -0700 Subject: [PATCH 3/3] test(blackbox): keep drift off stacks touched by a canceled command A canceled changeset's resources can stay registered as in-progress with the synchronizer past the command's terminal state, during which sync skips them and injected drift cannot absorb. The window is not observable from outside the agent, so stacks touched by a canceled command leave the drift-target pool for the rest of the iteration. --- tests/blackbox/executor.go | 9 +++++++++ tests/blackbox/state_model.go | 30 +++++++++++++++++++++++------- tests/blackbox/state_model_test.go | 20 ++++++++++++++++++++ 3 files changed, 52 insertions(+), 7 deletions(-) diff --git a/tests/blackbox/executor.go b/tests/blackbox/executor.go index ae3ba8b56..b120efc20 100644 --- a/tests/blackbox/executor.go +++ b/tests/blackbox/executor.go @@ -1470,6 +1470,15 @@ func (h *TestHarness) executeCancel(t *testing.T, op *Operation, model *StateMod t.Logf("[op %d] Cancel command %s → accepted", op.SequenceNum, target.CommandID) + // The canceled changeset's resources can stay registered as in-progress + // with the synchronizer for an unobservable time after the command reaches + // its terminal state, during which sync skips them. Drift injected on them + // could then never absorb, so take their stacks out of drift targeting for + // the rest of the iteration. + for _, ref := range target.RequestedSlots { + model.MarkDriftExcludedStack(ref.StackIndex) + } + // Wait for the canceled command to reach a terminal state so we can read // per-resource-update outcomes. The most recent accepted command may be // a stack-expirer destroy tracked by executeCheckTTL, which never diff --git a/tests/blackbox/state_model.go b/tests/blackbox/state_model.go index 99db09c9f..83c4eb8ea 100644 --- a/tests/blackbox/state_model.go +++ b/tests/blackbox/state_model.go @@ -72,6 +72,12 @@ type StateModel struct { // Populated from command response ResourceUpdate.NativeID on successful // creates/updates. Cleared on successful deletes. NativeIDs map[string]string + // DriftExcludedStacks marks stacks whose resources are no longer valid + // drift targets this iteration: a canceled changeset can leave its + // resources registered as in-progress with the synchronizer for an + // unobservable time, during which sync skips them and drift on them + // cannot absorb deterministically. + DriftExcludedStacks map[int]bool } // NewStateModel creates a state model with the given number of stacks, @@ -120,16 +126,23 @@ func NewStateModel(stackCount, resourcesPerStack int) *StateModel { providerLabel = stacks[0].Label } return &StateModel{ - Stacks: stacks, - ResourcesPerStack: resourcesPerStack, - Pool: pool, - ProviderStackLabel: providerLabel, - UnmanagedResources: make(map[string]*ExpectedUnmanagedResource), - AuthoritativeSlots: make(map[string]bool), - NativeIDs: make(map[string]string), + Stacks: stacks, + ResourcesPerStack: resourcesPerStack, + Pool: pool, + ProviderStackLabel: providerLabel, + UnmanagedResources: make(map[string]*ExpectedUnmanagedResource), + AuthoritativeSlots: make(map[string]bool), + NativeIDs: make(map[string]string), + DriftExcludedStacks: make(map[int]bool), } } +// MarkDriftExcludedStack removes a stack's resources from drift targeting for +// the rest of the iteration. See DriftExcludedStacks. +func (m *StateModel) MarkDriftExcludedStack(stackIdx int) { + m.DriftExcludedStacks[stackIdx] = true +} + func slotKeyString(stackIdx, slotIdx int) string { return fmt.Sprintf("%d/%d", stackIdx, slotIdx) } @@ -171,6 +184,9 @@ func (m *StateModel) FindDriftEligibleResource(sequenceNum int) (stackIdx, slotI busyStacks[ref.StackIndex] = true } } + for stackIdx := range m.DriftExcludedStacks { + busyStacks[stackIdx] = true + } type candidate struct { stackIdx, slotIdx int diff --git a/tests/blackbox/state_model_test.go b/tests/blackbox/state_model_test.go index d72098c66..cae423d43 100644 --- a/tests/blackbox/state_model_test.go +++ b/tests/blackbox/state_model_test.go @@ -483,6 +483,26 @@ func TestStateModel_DriftEligibilityCrossStackNeedsQuiescentProvider(t *testing. assert.Equal(t, "test-9", nativeID) } +// A canceled changeset can leave its resources registered as in-progress with +// the synchronizer for an unobservable time, during which sync skips them and +// drift on them cannot absorb. Stacks touched by a canceled command are +// excluded from drift targeting for the rest of the iteration. +func TestStateModel_DriftEligibilitySkipsCancelPoisonedStacks(t *testing.T) { + model := NewStateModel(2, 3) + model.ApplyCreated(0, []int{0}, `{"Name":"res-stack-0-a","Value":"v1"}`) + model.SetNativeID(0, 0, "test-1") + model.ApplyCreated(1, []int{0}, `{"Name":"res-stack-1-a","Value":"v1"}`) + model.SetNativeID(1, 0, "test-2") + + model.MarkDriftExcludedStack(1) + + for seq := range 4 { + stackIdx, _, _, _, _, _, ok := model.FindDriftEligibleResource(seq) + require.True(t, ok) + assert.Equal(t, 0, stackIdx, "stack 1 was touched by a canceled command and must be excluded") + } +} + // Eligible drift targets must be enumerated in a stable order: selection is // keyed off the generated sequence number, and map-iteration order would make // the picked resource nondeterministic across runs.