diff --git a/changes/45635-windows-batched-reconciler b/changes/45635-windows-batched-reconciler new file mode 100644 index 0000000000..8fc49c1db1 --- /dev/null +++ b/changes/45635-windows-batched-reconciler @@ -0,0 +1 @@ +- Improved Windows MDM configuration profile performance. Changes to Windows profiles now reach hosts more quickly. Large changes that affect many hosts at once, such as adding or removing profiles across a team or transferring many hosts between teams, now finish faster and put significantly less load on Fleet's database, keeping the server responsive at scale. diff --git a/server/datastore/mysql/microsoft_mdm_batched.go b/server/datastore/mysql/microsoft_mdm_batched.go new file mode 100644 index 0000000000..7a06078d9b --- /dev/null +++ b/server/datastore/mysql/microsoft_mdm_batched.go @@ -0,0 +1,327 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + + "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" + "github.com/fleetdm/fleet/v4/server/fleet" + common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql" + "github.com/jmoiron/sqlx" +) + +// listWindowsMDMHostsForReconcileBatchTransaction returns up to batchSize Windows-MDM-enrolled hosts with uuid > afterHostUUID, +// ordered ascending by uuid, along with the fields the batched reconciler needs to compute the desired state in memory. +// +// platform 'windows', an mdm_windows_enrollments row, and a host_mdm row with enrolled = 1. The two enrollment relationships are +// expressed as EXISTS subqueries rather than JOINs so a host with more than one mdm_windows_enrollments row (the table has no +// uniqueness on host_uuid) yields exactly one host record here. +func (ds *Datastore) listWindowsMDMHostsForReconcileBatchTransaction( + ctx context.Context, + tx common_mysql.DBReadTx, + afterHostUUID string, + batchSize int, +) ([]*fleet.WindowsHostReconcileInfo, error) { + const stmt = ` + SELECT + h.id AS id, + h.uuid AS uuid, + h.team_id AS team_id, + h.label_updated_at AS label_updated_at + FROM hosts h + WHERE + h.platform = 'windows' + AND h.uuid > ? + AND EXISTS ( + SELECT 1 FROM mdm_windows_enrollments mwe WHERE mwe.host_uuid = h.uuid + ) + AND EXISTS ( + SELECT 1 FROM host_mdm hmdm WHERE hmdm.host_id = h.id AND hmdm.enrolled = 1 + ) + ORDER BY h.uuid + LIMIT ? + ` + + var hosts []*fleet.WindowsHostReconcileInfo + if err := sqlx.SelectContext(ctx, tx, &hosts, stmt, afterHostUUID, batchSize); err != nil { + return nil, ctxerr.Wrap(ctx, err, "list windows mdm hosts for reconcile batch") + } + return hosts, nil +} + +// listWindowsProfilesForReconcileTransaction loads every Windows configuration profile in the system, paired with its label +// assignments. Mirrors the Apple listAppleProfilesForReconcileTransaction so the in-memory handlers apply the same "broken-label" +// semantics and broken profiles are exempted from removal. +func (ds *Datastore) listWindowsProfilesForReconcileTransaction( + ctx context.Context, + tx common_mysql.DBReadTx, +) ([]*fleet.WindowsProfileForReconcile, error) { + type profileRow struct { + ProfileUUID string `db:"profile_uuid"` + ProfileName string `db:"name"` + TeamID uint `db:"team_id"` + Checksum []byte `db:"checksum"` + SecretsUpdatedAt sql.NullTime `db:"secrets_updated_at"` + } + + const profStmt = ` + SELECT profile_uuid, name, team_id, checksum, secrets_updated_at + FROM mdm_windows_configuration_profiles + ` + + var rows []profileRow + if err := sqlx.SelectContext(ctx, tx, &rows, profStmt); err != nil { + return nil, ctxerr.Wrap(ctx, err, "list windows profiles for reconcile") + } + if len(rows) == 0 { + return nil, nil + } + + byUUID := make(map[string]*fleet.WindowsProfileForReconcile, len(rows)) + out := make([]*fleet.WindowsProfileForReconcile, 0, len(rows)) + for _, r := range rows { + p := &fleet.WindowsProfileForReconcile{ + ProfileUUID: r.ProfileUUID, + ProfileName: r.ProfileName, + TeamID: r.TeamID, + Checksum: r.Checksum, + } + if r.SecretsUpdatedAt.Valid { + t := r.SecretsUpdatedAt.Time + p.SecretsUpdatedAt = &t + } + byUUID[r.ProfileUUID] = p + out = append(out, p) + } + + // Load label assignments, joining labels to get membership type and label creation time (needed by the exclude-any handler). + // Broken labels (label_id IS NULL after the LEFT JOIN, i.e. the label was deleted) are retained so the handlers can + // disqualify/exempt the profile. + // + // Leave created_at un-COALESCE'd. NULL here means a broken (deleted) label and is intentional. + const labelStmt = ` + SELECT + mcpl.windows_profile_uuid AS profile_uuid, + mcpl.label_id AS label_id, + mcpl.exclude AS exclude, + mcpl.require_all AS require_all, + lbl.created_at AS label_created_at, + COALESCE(lbl.label_membership_type, 0) AS label_membership_type + FROM mdm_configuration_profile_labels mcpl + LEFT JOIN labels lbl ON lbl.id = mcpl.label_id + WHERE mcpl.windows_profile_uuid IS NOT NULL + ` + + type labelRow struct { + ProfileUUID string `db:"profile_uuid"` + LabelID sql.NullInt64 `db:"label_id"` + Exclude bool `db:"exclude"` + RequireAll bool `db:"require_all"` + LabelCreatedAt sql.NullTime `db:"label_created_at"` + LabelMembershipType int `db:"label_membership_type"` + } + + var labelRows []labelRow + if err := sqlx.SelectContext(ctx, tx, &labelRows, labelStmt); err != nil { + return nil, ctxerr.Wrap(ctx, err, "list windows profile labels for reconcile") + } + + // Per-profile include-mode discovery. Include labels for a single profile must share a single require_all value; the first + // include row sets the mode and later disagreements mark it mixed. Exclude rows always go to ExcludeLabels and have a single + // "exclude any" semantic. A profile may carry both an include set and an exclude set. + type includeAccum struct { + set bool + mode fleet.MDMProfileIncludeMode + mixed bool + } + includeModes := make(map[string]*includeAccum, len(byUUID)) + + for _, lr := range labelRows { + p, ok := byUUID[lr.ProfileUUID] + if !ok { + continue + } + + ref := fleet.MDMProfileLabelRef{ + LabelMembershipType: lr.LabelMembershipType, + } + if lr.LabelID.Valid { + id := uint(lr.LabelID.Int64) //nolint:gosec // dismiss G115: labels.id is int unsigned in MySQL + ref.LabelID = &id + } + if lr.LabelCreatedAt.Valid { + ref.CreatedAt = lr.LabelCreatedAt.Time + } + + if lr.Exclude { + p.ExcludeLabels = append(p.ExcludeLabels, ref) + continue + } + + // Include row. + p.IncludeLabels = append(p.IncludeLabels, ref) + + rowMode := fleet.MDMProfileIncludeAny + if lr.RequireAll { + rowMode = fleet.MDMProfileIncludeAll + } + + ia := includeModes[lr.ProfileUUID] + if ia == nil { + ia = &includeAccum{} + includeModes[lr.ProfileUUID] = ia + } + if !ia.set { + ia.mode = rowMode + ia.set = true + } else if ia.mode != rowMode { + ia.mixed = true + } + } + + for uuid, ia := range includeModes { + p := byUUID[uuid] + if p == nil { + // Unreachable: every includeModes key came from a label row whose profile UUID is in byUUID. Guard anyway to satisfy nil + // analysis. + continue + } + if ia.mixed { + // Defensive: include rows disagreed on require_all (should be impossible in production since the upsert path enforces a single mode). + // Drop the include set so we don't guess at intent; exclude labels (if any) are preserved. + p.IncludeLabels = nil + p.IncludeMode = fleet.MDMProfileIncludeNone + errMsg := "windows profile has mixed include label modes; ignoring include labels" + ds.logger.ErrorContext(ctx, errMsg, "profile_uuid", uuid, "team_id", + p.TeamID) + ctxerr.Handle(ctx, errors.New(errMsg)) + continue + } + p.IncludeMode = ia.mode + } + + return out, nil +} + +// bulkGetHostMDMWindowsProfilesByUUIDsTransaction returns the current host_mdm_windows_profiles rows for the given host UUIDs, +// grouped by host UUID. +// +// The caller (GetWindowsProfileReconcileSnapshot) always passes the reconcile host window, bounded by +// reconcileWindowsProfilesBatchSize (a per-tick read budget in the low thousands), which stays far under MySQL's ~65k +// prepared-statement placeholder limit. The IN clause therefore fits in a single query and is intentionally not batched. +func (ds *Datastore) bulkGetHostMDMWindowsProfilesByUUIDsTransaction( + ctx context.Context, + tx common_mysql.DBReadTx, + hostUUIDs []string, +) (map[string][]*fleet.MDMWindowsProfilePayload, error) { + out := make(map[string][]*fleet.MDMWindowsProfilePayload, len(hostUUIDs)) + if len(hostUUIDs) == 0 { + return out, nil + } + + const stmt = ` + SELECT + profile_uuid, + host_uuid, + profile_name, + status, + operation_type, + COALESCE(detail, '') AS detail, + command_uuid, + retries, + checksum, + secrets_updated_at + FROM host_mdm_windows_profiles + WHERE host_uuid IN (?) + ` + + q, args, err := sqlx.In(stmt, hostUUIDs) + if err != nil { + return nil, ctxerr.Wrap(ctx, err, "build host mdm windows profiles query") + } + + var rows []*fleet.MDMWindowsProfilePayload + if err := sqlx.SelectContext(ctx, tx, &rows, q, args...); err != nil { + return nil, ctxerr.Wrap(ctx, err, "select host mdm windows profiles") + } + + for _, r := range rows { + out[r.HostUUID] = append(out[r.HostUUID], r) + } + + return out, nil +} + +// GetWindowsProfileReconcileSnapshot loads the four pieces of state the batched Windows profile reconciler needs — the bounded +// host window, every profile (with label assignments), host↔label memberships restricted to labels referenced by those profiles, +// and current host_mdm_windows_profiles rows for the hosts in the window. All reads run inside a single read-only transaction so +// they observe one MySQL snapshot. +// +// The read-only REPEATABLE READ transaction is load-bearing, not incidental: it makes the desired-state inputs (profiles, labels, +// memberships) and the current state (host_mdm_windows_profiles) coherent at one instant, so a concurrent admin mutation (e.g. +// deleting a profile, which also deletes its host rows) cannot produce a torn diff with spurious install/remove targets. Do not +// pull these reads out of the transaction (e.g. to load profiles once per tick) without weighing that consistency loss. +// +// When the host window is empty the remaining queries are skipped — the caller short-circuits in that case anyway, and there's no +// point loading profiles or memberships we won't use. Mirrors GetAppleProfileReconcileSnapshot. +func (ds *Datastore) GetWindowsProfileReconcileSnapshot(ctx context.Context, afterHostUUID string, batchSize int) ( + hosts []*fleet.WindowsHostReconcileInfo, + allProfiles []*fleet.WindowsProfileForReconcile, + hostLabels map[uint]map[uint]struct{}, + currentByHost map[string][]*fleet.MDMWindowsProfilePayload, + err error, +) { + err = ds.withReadTx(ctx, func(tx common_mysql.DBReadTx) error { + var inner error + hosts, inner = ds.listWindowsMDMHostsForReconcileBatchTransaction(ctx, tx, afterHostUUID, batchSize) + if inner != nil { + return inner + } + if len(hosts) == 0 { + return nil + } + + allProfiles, inner = ds.listWindowsProfilesForReconcileTransaction(ctx, tx) + if inner != nil { + return inner + } + + hostIDs := make([]uint, 0, len(hosts)) + hostUUIDs := make([]string, 0, len(hosts)) + for _, h := range hosts { + hostIDs = append(hostIDs, h.HostID) + hostUUIDs = append(hostUUIDs, h.UUID) + } + + labelIDSet := make(map[uint]struct{}) + for _, p := range allProfiles { + for _, lr := range p.IncludeLabels { + if lr.LabelID != nil { + labelIDSet[*lr.LabelID] = struct{}{} + } + } + for _, lr := range p.ExcludeLabels { + if lr.LabelID != nil { + labelIDSet[*lr.LabelID] = struct{}{} + } + } + } + labelIDs := make([]uint, 0, len(labelIDSet)) + for id := range labelIDSet { + labelIDs = append(labelIDs, id) + } + + hostLabels, inner = ds.bulkGetHostLabelMembershipsTransaction(ctx, tx, hostIDs, labelIDs) + if inner != nil { + return inner + } + + currentByHost, inner = ds.bulkGetHostMDMWindowsProfilesByUUIDsTransaction(ctx, tx, hostUUIDs) + return inner + }) + if err != nil { + return nil, nil, nil, nil, ctxerr.Wrap(ctx, err, "windows profile reconcile snapshot") + } + return hosts, allProfiles, hostLabels, currentByHost, nil +} diff --git a/server/fleet/datastore.go b/server/fleet/datastore.go index 8dfed4bc96..80799dfe63 100644 --- a/server/fleet/datastore.go +++ b/server/fleet/datastore.go @@ -2329,6 +2329,19 @@ type Datastore interface { // the Windows MDM reconciliation cron. See GetMDMWindowsReconcileCursor. SetMDMWindowsReconcileCursor(ctx context.Context, cursor string) error + // GetWindowsProfileReconcileSnapshot returns a consistent snapshot of the state needed by the batched Windows profile reconciler: + // the bounded host window (afterHostUUID, batchSize), every Windows profile with its label assignments, host↔label memberships + // for labels referenced by those profiles, and current host_mdm_windows_profiles rows for the host window. All reads run inside a + // single read-only MySQL transaction so they observe one snapshot. If the host window is empty the remaining slices and maps are + // nil. See ReconcileWindowsProfiles. + GetWindowsProfileReconcileSnapshot(ctx context.Context, afterHostUUID string, batchSize int) ( + hosts []*WindowsHostReconcileInfo, + allProfiles []*WindowsProfileForReconcile, + hostLabels map[uint]map[uint]struct{}, + currentByHost map[string][]*MDMWindowsProfilePayload, + err error, + ) + // GetAppleMDMHostForReconcile returns reconcile info for a single Apple // MDM-enrolled host UUID, or (nil, nil) if the host is not enrolled or // not an Apple platform. Used by ReconcileAppleProfilesForEnrollingHost (the diff --git a/server/fleet/windows_mdm.go b/server/fleet/windows_mdm.go index ed2c349b5a..d557701bf2 100644 --- a/server/fleet/windows_mdm.go +++ b/server/fleet/windows_mdm.go @@ -585,6 +585,57 @@ type MDMWindowsProfileContents struct { Checksum []byte `db:"checksum"` } +// WindowsHostReconcileInfo is a per-host record used by the batched Windows profile reconciler. It contains only the fields +// needed to decide which profiles should be installed on the host given its team and label membership. Mirrors +// AppleHostReconcileInfo +type WindowsHostReconcileInfo struct { + HostID uint `db:"id"` + UUID string `db:"uuid"` + TeamID *uint `db:"team_id"` + LabelUpdatedAt time.Time `db:"label_updated_at"` +} + +// EffectiveTeamID returns 0 for hosts not in a team. team_id=0 is its own team (the "no team" / global scope). Equality between +// EffectiveTeamID and a profile's team_id is the correct match check. See AppleHostReconcileInfo.EffectiveTeamID. +func (h *WindowsHostReconcileInfo) EffectiveTeamID() uint { + if h.TeamID == nil { + return 0 + } + return *h.TeamID +} + +// WindowsProfileForReconcile is the profile data needed by the batched Windows reconciler to compute desired state per host in +// memory. The label-gating fields mirror AppleProfileForReconcile exactly so the same shared dispatcher and handlers +// (server/mdm/reconcile) run against both platforms. +// +// Include and exclude labels are stored separately so a profile can carry both: applicability becomes (include gate passes) AND +// (exclude gate passes), with each gate skipped when its slice is empty. +type WindowsProfileForReconcile struct { + ProfileUUID string + ProfileName string + TeamID uint // 0 means global + Checksum []byte + SecretsUpdatedAt *time.Time + IncludeMode MDMProfileIncludeMode + IncludeLabels []MDMProfileLabelRef + ExcludeLabels []MDMProfileLabelRef +} + +func (p *WindowsProfileForReconcile) GetTeamID() uint { return p.TeamID } +func (p *WindowsProfileForReconcile) GetIncludeMode() MDMProfileIncludeMode { return p.IncludeMode } +func (p *WindowsProfileForReconcile) GetIncludeLabels() []MDMProfileLabelRef { + return p.IncludeLabels +} +func (p *WindowsProfileForReconcile) GetExcludeLabels() []MDMProfileLabelRef { + return p.ExcludeLabels +} + +// HasBrokenLabel reports whether any include or exclude label on the profile references a deleted label. Used to keep +// broken-label profiles exempt from removal. See AppleProfileForReconcile.HasBrokenLabel. +func (p *WindowsProfileForReconcile) HasBrokenLabel() bool { + return anyMDMLabelBroken(p.IncludeLabels) || anyMDMLabelBroken(p.ExcludeLabels) +} + // MDMWindowsWipeType specifies what type of remote wipe we want // to perform. type MDMWindowsWipeType int diff --git a/server/mdm/microsoft/reconcile.go b/server/mdm/microsoft/reconcile.go new file mode 100644 index 0000000000..1d6f5c3b51 --- /dev/null +++ b/server/mdm/microsoft/reconcile.go @@ -0,0 +1,116 @@ +package microsoft_mdm + +import ( + "bytes" + + "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/fleetdm/fleet/v4/server/mdm/reconcile" +) + +// ComputeWindowsReconcileDeltas evaluates the desired profile state for each host in the input set using the SHARED dispatcher +// (server/mdm/reconcile), then diffs against current host_mdm_windows_profiles rows to produce install and remove sets. +// +// profilesByTeam groups every loaded profile by its team_id; profilesWithBrokenLabels holds the UUIDs of profiles carrying at +// least one deleted label (kept out of removal). +func ComputeWindowsReconcileDeltas( + hosts []*fleet.WindowsHostReconcileInfo, + hostLabels map[uint]map[uint]struct{}, + currentByHost map[string][]*fleet.MDMWindowsProfilePayload, + profilesByTeam map[uint][]*fleet.WindowsProfileForReconcile, + profilesWithBrokenLabels map[string]struct{}, +) (toInstall, toRemove []*fleet.MDMWindowsProfilePayload) { + for _, host := range hosts { + teamProfiles := profilesByTeam[host.EffectiveTeamID()] + desired := make(map[string]*fleet.WindowsProfileForReconcile, len(teamProfiles)) + + labelsForHost := hostLabels[host.HostID] + + for _, p := range teamProfiles { + // Determine if this profile should be on this host + if !reconcile.EntityAppliesToHost(p, host.EffectiveTeamID(), host.LabelUpdatedAt, labelsForHost) { + continue + } + desired[p.ProfileUUID] = p + } + + current := currentByHost[host.UUID] + currentByProfile := make(map[string]*fleet.MDMWindowsProfilePayload, len(current)) + for _, c := range current { + currentByProfile[c.ProfileUUID] = c + } + + // Install set + for profUUID, p := range desired { + c, present := currentByProfile[profUUID] + needsInstall := false + switch { + case !present: + // profile in desired (A) but not in current (B). + needsInstall = true + case !bytes.Equal(c.Checksum, p.Checksum): + // profile content changed (hmwp.checksum != ds.checksum). + needsInstall = true + case p.SecretsUpdatedAt != nil && c.SecretsUpdatedAt != nil && c.SecretsUpdatedAt.Before(*p.SecretsUpdatedAt): + // secret variables updated. Matches + // IFNULL(hmwp.secrets_updated_at < ds.secrets_updated_at, FALSE): + // only fires when BOTH timestamps are present and current is older. + needsInstall = true + case c.OperationType == fleet.MDMOperationTypeInstall && c.Status == nil: + // install was never sent (NULL status); re-push. + needsInstall = true + case c.OperationType == fleet.MDMOperationTypeRemove && !isTerminalRemoveStatus(c.Status): + // currently marked for removal but not an in-flight or completed + // removal — flip back to install. Matches + // operation_type = remove AND COALESCE(status,'') NOT IN ('verifying','verified'). + needsInstall = true + } + if !needsInstall { + continue + } + + toInstall = append(toInstall, &fleet.MDMWindowsProfilePayload{ + ProfileUUID: p.ProfileUUID, + ProfileName: p.ProfileName, + HostUUID: host.UUID, + Checksum: p.Checksum, + SecretsUpdatedAt: p.SecretsUpdatedAt, + }) + } + + // Remove set + for profUUID, c := range currentByProfile { + if _, stillDesired := desired[profUUID]; stillDesired { + continue + } + // Skip rows already processing a remove + if c.OperationType == fleet.MDMOperationTypeRemove && c.Status != nil { + continue + } + // Keep (don't remove) profiles with a broken label + if _, broken := profilesWithBrokenLabels[profUUID]; broken { + continue + } + + toRemove = append(toRemove, &fleet.MDMWindowsProfilePayload{ + ProfileUUID: c.ProfileUUID, + ProfileName: c.ProfileName, + HostUUID: host.UUID, + OperationType: c.OperationType, + Detail: c.Detail, + Status: c.Status, + CommandUUID: c.CommandUUID, + }) + } + } + return toInstall, toRemove +} + +// isTerminalRemoveStatus reports whether a remove row's status is one that the install query treats as "leave alone" +// (verifying/verified). A NULL status, or any other status (e.g. pending, failed), means the remove can be flipped back to +// install. +func isTerminalRemoveStatus(status *fleet.MDMDeliveryStatus) bool { + if status == nil { + return false + } + return *status == fleet.MDMDeliveryVerifying || *status == fleet.MDMDeliveryVerified +} diff --git a/server/mdm/microsoft/reconcile_test.go b/server/mdm/microsoft/reconcile_test.go new file mode 100644 index 0000000000..8a722126cb --- /dev/null +++ b/server/mdm/microsoft/reconcile_test.go @@ -0,0 +1,433 @@ +package microsoft_mdm + +import ( + "testing" + "time" + + "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/stretchr/testify/require" +) + +// deltaSet collapses a delta slice (install or remove) into a set of "hostUUID|profileUUID" keys so assertions don't depend on +// slice order. +func deltaSet(payloads []*fleet.MDMWindowsProfilePayload) map[string]struct{} { + out := make(map[string]struct{}, len(payloads)) + for _, p := range payloads { + out[p.HostUUID+"|"+p.ProfileUUID] = struct{}{} + } + return out +} + +func key(hostUUID, profileUUID string) string { return hostUUID + "|" + profileUUID } + +// TestComputeWindowsReconcileDeltasInstallRules covers install scenarios. The host is a no-team host with one global, label-less +// profile; only the current host_mdm_windows_profiles row varies between cases. +func TestComputeWindowsReconcileDeltasInstallRules(t *testing.T) { + host := &fleet.WindowsHostReconcileInfo{HostID: 1, UUID: "h1", TeamID: nil} + desiredChecksum := []byte("checksum-A") + newer := time.Now() + older := newer.Add(-time.Hour) + + profile := &fleet.WindowsProfileForReconcile{ + ProfileUUID: "p1", + ProfileName: "Profile 1", + TeamID: 0, + Checksum: desiredChecksum, + } + + cases := []struct { + name string + current *fleet.MDMWindowsProfilePayload // nil => no current row + profileMod func(p *fleet.WindowsProfileForReconcile) + wantInstall bool + }{ + { + name: "no current row installs", + current: nil, + wantInstall: true, + }, + { + name: "matching install row does not reinstall", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified)}, + wantInstall: false, + }, + { + name: "checksum mismatch installs", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: []byte("stale"), OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified)}, + wantInstall: true, + }, + { + name: "secrets updated (both present, current older) installs", + profileMod: func(p *fleet.WindowsProfileForReconcile) { + p.SecretsUpdatedAt = &newer + }, + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified), SecretsUpdatedAt: &older}, + wantInstall: true, + }, + { + name: "desired has secrets but current has none does NOT install (IFNULL=FALSE)", + profileMod: func(p *fleet.WindowsProfileForReconcile) { + p.SecretsUpdatedAt = &newer + }, + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified), SecretsUpdatedAt: nil}, + wantInstall: false, + }, + { + name: "install op with NULL status reinstalls", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeInstall, Status: nil}, + wantInstall: true, + }, + { + name: "install op pending does not reinstall", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryPending)}, + wantInstall: false, + }, + { + name: "remove op NULL status flips back to install", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeRemove, Status: nil}, + wantInstall: true, + }, + { + name: "remove op pending flips back to install", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeRemove, Status: new(fleet.MDMDeliveryPending)}, + wantInstall: true, + }, + { + name: "remove op verifying does not flip back", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeRemove, Status: new(fleet.MDMDeliveryVerifying)}, + wantInstall: false, + }, + { + name: "remove op verified does not flip back", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "p1", HostUUID: "h1", Checksum: desiredChecksum, OperationType: fleet.MDMOperationTypeRemove, Status: new(fleet.MDMDeliveryVerified)}, + wantInstall: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + p := *profile + if tc.profileMod != nil { + tc.profileMod(&p) + } + currentByHost := map[string][]*fleet.MDMWindowsProfilePayload{} + if tc.current != nil { + currentByHost["h1"] = []*fleet.MDMWindowsProfilePayload{tc.current} + } + + toInstall, toRemove := ComputeWindowsReconcileDeltas( + []*fleet.WindowsHostReconcileInfo{host}, + nil, + currentByHost, + map[uint][]*fleet.WindowsProfileForReconcile{0: {&p}}, + nil, + ) + require.Empty(t, toRemove) + if tc.wantInstall { + require.Contains(t, deltaSet(toInstall), key("h1", "p1")) + // install payload carries the desired profile's content. + require.Equal(t, p.Checksum, toInstall[0].Checksum) + } else { + require.Empty(t, toInstall) + } + }) + } +} + +// TestComputeWindowsReconcileDeltasRemoveRules covers the remove scenarios: current rows with no desired-state match are removed, +// except rows already processing a remove and except broken-label profiles. +func TestComputeWindowsReconcileDeltasRemoveRules(t *testing.T) { + host := &fleet.WindowsHostReconcileInfo{HostID: 1, UUID: "h1", TeamID: nil} + + cases := []struct { + name string + current *fleet.MDMWindowsProfilePayload + broken bool + wantRemove bool + }{ + { + name: "current install not desired is removed", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "gone", HostUUID: "h1", ProfileName: "Gone", OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified), Detail: "prior detail", CommandUUID: "cmd-1"}, + wantRemove: true, + }, + { + name: "remove op with NULL status is (re)removed", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "gone", HostUUID: "h1", OperationType: fleet.MDMOperationTypeRemove, Status: nil}, + wantRemove: true, + }, + { + name: "remove op already in-flight is skipped", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "gone", HostUUID: "h1", OperationType: fleet.MDMOperationTypeRemove, Status: new(fleet.MDMDeliveryPending)}, + wantRemove: false, + }, + { + name: "broken-label profile is kept (not removed)", + current: &fleet.MDMWindowsProfilePayload{ProfileUUID: "gone", HostUUID: "h1", OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified)}, + broken: true, + wantRemove: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var broken map[string]struct{} + if tc.broken { + broken = map[string]struct{}{"gone": {}} + } + toInstall, toRemove := ComputeWindowsReconcileDeltas( + []*fleet.WindowsHostReconcileInfo{host}, + nil, + map[string][]*fleet.MDMWindowsProfilePayload{"h1": {tc.current}}, + nil, // no desired profiles + broken, + ) + require.Empty(t, toInstall) + if tc.wantRemove { + require.Len(t, toRemove, 1) + got := toRemove[0] + require.Equal(t, "gone", got.ProfileUUID) + require.Equal(t, "h1", got.HostUUID) + // Remove payloads carry these fields straight from the current row, matching the legacy remove query's SELECT list. + require.Equal(t, tc.current.ProfileName, got.ProfileName) + require.Equal(t, tc.current.OperationType, got.OperationType) + require.Equal(t, tc.current.Status, got.Status) + require.Equal(t, tc.current.Detail, got.Detail) + require.Equal(t, tc.current.CommandUUID, got.CommandUUID) + } else { + require.Empty(t, toRemove) + } + }) + } +} + +// TestComputeWindowsReconcileDeltasTeamGating verifies that a host only receives profiles for its own team; team_id=0 is its own +// scope, a teamed host does not inherit global profiles. +func TestComputeWindowsReconcileDeltasTeamGating(t *testing.T) { + globalProfile := &fleet.WindowsProfileForReconcile{ProfileUUID: "pg", ProfileName: "global", TeamID: 0, Checksum: []byte("c")} + teamProfile := &fleet.WindowsProfileForReconcile{ProfileUUID: "pt", ProfileName: "team", TeamID: 5, Checksum: []byte("c")} + + noTeamHost := &fleet.WindowsHostReconcileInfo{HostID: 1, UUID: "h-global", TeamID: nil} + teamedHost := &fleet.WindowsHostReconcileInfo{HostID: 2, UUID: "h-team", TeamID: new(uint(5))} + + profilesByTeam := map[uint][]*fleet.WindowsProfileForReconcile{ + 0: {globalProfile}, + 5: {teamProfile}, + } + + toInstall, toRemove := ComputeWindowsReconcileDeltas( + []*fleet.WindowsHostReconcileInfo{noTeamHost, teamedHost}, + nil, + map[string][]*fleet.MDMWindowsProfilePayload{}, + profilesByTeam, + nil, + ) + require.Empty(t, toRemove) + + got := deltaSet(toInstall) + require.Contains(t, got, key("h-global", "pg")) + require.Contains(t, got, key("h-team", "pt")) + require.NotContains(t, got, key("h-global", "pt")) + require.NotContains(t, got, key("h-team", "pg")) + require.Len(t, got, 2) +} + +// TestComputeWindowsReconcileDeltasLabelMatrix confirms the compute routes the label gates through the shared dispatcher: +// include-all, include-any, exclude-any, combined include+exclude, broken labels, and dynamic-label timing. The handlers +// themselves are unit-tested in server/mdm/reconcile; here we assert the desired-state membership for representative cases. +func TestComputeWindowsReconcileDeltasLabelMatrix(t *testing.T) { + hostLabels := map[uint]map[uint]struct{}{ + 1: {10: {}, 11: {}}, // host 1 is a member of labels 10 and 11 + } + host := &fleet.WindowsHostReconcileInfo{HostID: 1, UUID: "h1", TeamID: nil, LabelUpdatedAt: time.Now()} + oldLabel := time.Now().Add(-24 * time.Hour) + + cases := []struct { + name string + profile *fleet.WindowsProfileForReconcile + wantInstall bool + }{ + { + name: "include-all member of all installs", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAll, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}, {LabelID: new(uint(11))}}, + }, + wantInstall: true, + }, + { + name: "include-all missing one does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAll, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}, {LabelID: new(uint(99))}}, + }, + wantInstall: false, + }, + { + name: "include-all with broken label does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAll, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}, {LabelID: nil}}, + }, + wantInstall: false, + }, + { + name: "include-any member of one installs", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAny, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}, {LabelID: new(uint(99))}}, + }, + wantInstall: true, + }, + { + name: "include-any member of none does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAny, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(98))}, {LabelID: new(uint(99))}}, + }, + wantInstall: false, + }, + { + name: "exclude-any non-member installs", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(99))}}, + }, + wantInstall: true, + }, + { + name: "exclude-any member does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}}, + }, + wantInstall: false, + }, + { + name: "exclude-any broken label does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: nil}}, + }, + wantInstall: false, + }, + { + name: "include-all + exclude-any: in include, not in exclude installs", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAll, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}}, + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(99))}}, + }, + wantInstall: true, + }, + { + name: "include-all + exclude-any: in include AND in exclude does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAll, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}}, + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(11))}}, + }, + wantInstall: false, + }, + { + name: "exclude-any dynamic label created after host scan disqualifies", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + // host is NOT a member of label 50, but the dynamic label was created after the host's last label scan, so results are not yet + // reported and the host is treated as excluded. + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(50)), CreatedAt: time.Now().Add(time.Hour), LabelMembershipType: int(fleet.LabelMembershipTypeDynamic)}}, + }, + wantInstall: false, + }, + { + name: "exclude-any dynamic label created before host scan passes", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(50)), CreatedAt: oldLabel, LabelMembershipType: int(fleet.LabelMembershipTypeDynamic)}}, + }, + wantInstall: true, + }, + { + name: "include-any + exclude-any: in an include, not in exclude installs", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAny, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}, {LabelID: new(uint(99))}}, + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(98))}}, + }, + wantInstall: true, + }, + { + name: "include-any + exclude-any: in an include AND in exclude does not install", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + IncludeMode: fleet.MDMProfileIncludeAny, + IncludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(10))}}, + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(11))}}, + }, + wantInstall: false, + }, + { + // Deliberate cross-platform decision (see server/mdm/reconcile): the exclude-any timing safeguard applies only to dynamic labels. + // A host_vitals exclude label created after the host's last scan does NOT disqualify, so the profile still installs, unlike a + // dynamic label in the same situation above. + name: "exclude-any host_vitals label created after host scan still installs", + profile: &fleet.WindowsProfileForReconcile{ProfileUUID: "p", TeamID: 0, Checksum: []byte("c"), + ExcludeLabels: []fleet.MDMProfileLabelRef{{LabelID: new(uint(50)), CreatedAt: time.Now().Add(time.Hour), LabelMembershipType: int(fleet.LabelMembershipTypeHostVitals)}}, + }, + wantInstall: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + toInstall, toRemove := ComputeWindowsReconcileDeltas( + []*fleet.WindowsHostReconcileInfo{host}, + hostLabels, + map[string][]*fleet.MDMWindowsProfilePayload{}, + map[uint][]*fleet.WindowsProfileForReconcile{0: {tc.profile}}, + nil, + ) + require.Empty(t, toRemove) + if tc.wantInstall { + require.Contains(t, deltaSet(toInstall), key("h1", "p")) + } else { + require.Empty(t, toInstall) + } + }) + } +} + +// TestComputeWindowsReconcileDeltasMultipleProfilesPerHost exercises the per-host loop building install AND remove sets in a +// single call: one host whose team has a profile to install and a profile already in the desired/installed state (no-op), while +// also carrying a current row for a profile no longer desired (remove). This is the realistic shape the +// single-profile/single-direction cases above don't cover. +func TestComputeWindowsReconcileDeltasMultipleProfilesPerHost(t *testing.T) { + host := &fleet.WindowsHostReconcileInfo{HostID: 1, UUID: "h1", TeamID: nil} + checksum := []byte("c") + + profilesByTeam := map[uint][]*fleet.WindowsProfileForReconcile{ + 0: { + {ProfileUUID: "p-install", ProfileName: "Install", TeamID: 0, Checksum: checksum}, + {ProfileUUID: "p-noop", ProfileName: "NoOp", TeamID: 0, Checksum: checksum}, + }, + } + currentByHost := map[string][]*fleet.MDMWindowsProfilePayload{ + "h1": { + // already installed and matching -> neither install nor remove. + {ProfileUUID: "p-noop", HostUUID: "h1", Checksum: checksum, OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified)}, + // installed but no longer desired (not in profilesByTeam) -> remove. + {ProfileUUID: "p-remove", HostUUID: "h1", ProfileName: "Remove", Checksum: checksum, OperationType: fleet.MDMOperationTypeInstall, Status: new(fleet.MDMDeliveryVerified)}, + }, + } + + toInstall, toRemove := ComputeWindowsReconcileDeltas( + []*fleet.WindowsHostReconcileInfo{host}, + nil, + currentByHost, + profilesByTeam, + nil, + ) + + install := deltaSet(toInstall) + require.Len(t, install, 1) + require.Contains(t, install, key("h1", "p-install")) + require.NotContains(t, install, key("h1", "p-noop")) + + remove := deltaSet(toRemove) + require.Len(t, remove, 1) + require.Contains(t, remove, key("h1", "p-remove")) + require.NotContains(t, remove, key("h1", "p-noop")) +} diff --git a/server/mdm/reconcile/reconcile_test.go b/server/mdm/reconcile/reconcile_test.go index 7388604fc3..2bf2366a1f 100644 --- a/server/mdm/reconcile/reconcile_test.go +++ b/server/mdm/reconcile/reconcile_test.go @@ -21,7 +21,14 @@ func (e *testEntity) GetIncludeMode() fleet.MDMProfileIncludeMode { return e.in func (e *testEntity) GetIncludeLabels() []fleet.MDMProfileLabelRef { return e.includeLabels } func (e *testEntity) GetExcludeLabels() []fleet.MDMProfileLabelRef { return e.excludeLabels } func (e *testEntity) HasBrokenLabel() bool { - for _, l := range append(e.includeLabels, e.excludeLabels...) { + // Iterate the two slices separately: append(e.includeLabels, e.excludeLabels...) can write into e.includeLabels' backing array + // when it has spare capacity, leaking state across assertions. + for _, l := range e.includeLabels { + if l.LabelID == nil { + return true + } + } + for _, l := range e.excludeLabels { if l.LabelID == nil { return true } diff --git a/server/mock/datastore_mock.go b/server/mock/datastore_mock.go index 499d978ff4..def9924d53 100644 --- a/server/mock/datastore_mock.go +++ b/server/mock/datastore_mock.go @@ -1420,6 +1420,8 @@ type GetMDMWindowsReconcileCursorFunc func(ctx context.Context) (string, error) type SetMDMWindowsReconcileCursorFunc func(ctx context.Context, cursor string) error +type GetWindowsProfileReconcileSnapshotFunc func(ctx context.Context, afterHostUUID string, batchSize int) (hosts []*fleet.WindowsHostReconcileInfo, allProfiles []*fleet.WindowsProfileForReconcile, hostLabels map[uint]map[uint]struct{}, currentByHost map[string][]*fleet.MDMWindowsProfilePayload, err error) + type GetAppleMDMHostForReconcileFunc func(ctx context.Context, hostUUID string) (*fleet.AppleHostReconcileInfo, error) type ListAppleProfilesForReconcileByTeamFunc func(ctx context.Context, teamID uint) ([]*fleet.AppleProfileForReconcile, error) @@ -4191,6 +4193,9 @@ type DataStore struct { SetMDMWindowsReconcileCursorFunc SetMDMWindowsReconcileCursorFunc SetMDMWindowsReconcileCursorFuncInvoked bool + GetWindowsProfileReconcileSnapshotFunc GetWindowsProfileReconcileSnapshotFunc + GetWindowsProfileReconcileSnapshotFuncInvoked bool + GetAppleMDMHostForReconcileFunc GetAppleMDMHostForReconcileFunc GetAppleMDMHostForReconcileFuncInvoked bool @@ -10094,6 +10099,13 @@ func (s *DataStore) SetMDMWindowsReconcileCursor(ctx context.Context, cursor str return s.SetMDMWindowsReconcileCursorFunc(ctx, cursor) } +func (s *DataStore) GetWindowsProfileReconcileSnapshot(ctx context.Context, afterHostUUID string, batchSize int) (hosts []*fleet.WindowsHostReconcileInfo, allProfiles []*fleet.WindowsProfileForReconcile, hostLabels map[uint]map[uint]struct{}, currentByHost map[string][]*fleet.MDMWindowsProfilePayload, err error) { + s.mu.Lock() + s.GetWindowsProfileReconcileSnapshotFuncInvoked = true + s.mu.Unlock() + return s.GetWindowsProfileReconcileSnapshotFunc(ctx, afterHostUUID, batchSize) +} + func (s *DataStore) GetAppleMDMHostForReconcile(ctx context.Context, hostUUID string) (*fleet.AppleHostReconcileInfo, error) { s.mu.Lock() s.GetAppleMDMHostForReconcileFuncInvoked = true diff --git a/server/service/microsoft_mdm.go b/server/service/microsoft_mdm.go index 86e438f89a..a2fc262166 100644 --- a/server/service/microsoft_mdm.go +++ b/server/service/microsoft_mdm.go @@ -3460,21 +3460,38 @@ func (svc *Service) GetMDMWindowsProfilesSummary(ctx context.Context, teamID *ui return ps, nil } -// reconcileWindowsProfilesBatchSize bounds how many distinct hosts the -// Windows MDM reconciliation cron processes per tick. The cron uses a -// host_uuid cursor (persisted in Redis via the mysqlredis wrapper) to -// page through the pending-work universe in batches, smoothing the -// writer pressure that an unbounded reconciliation generates during -// bulk events like team transfers. +// reconcileWindowsProfilesBatchSize is the scan window: how many enrolled Windows hosts the reconciler reads per snapshot. +// Snapshot reads are cheap (indexed, no set-difference), so within a single tick the drain loop pages through many windows until +// a budget is hit. // -// var rather than const so property-based tests can shrink the batch size +// var rather than const so property-based tests can shrink the batch size. var reconcileWindowsProfilesBatchSize = 2000 +// reconcileWindowsProfilesDeliveryCap bounds how many distinct hosts the cron schedules for install/remove per tick. It governs +// the bulk case: once this many hosts have been delivered work, the tick stops even if scan budget remains, advancing the cursor +// only to the last delivered host so the remainder resumes next tick. This preserves the writer-pressure smoothing: a bulk change +// is spread across ~ceil(hosts/cap) ticks. Set <= 0 to disable the cap (drain the whole fleet, bounded only by the scan budget). +// +// var rather than const so tests can override it. +var reconcileWindowsProfilesDeliveryCap = 2000 + +// reconcileWindowsProfilesScanBudget is the wall-clock budget for a single tick's drain loop. It governs the sparse/idle case: a +// no-work pass over the whole fleet completes within one tick, collapsing single-change latency from ceil(hosts/batch) x interval +// to roughly the actual work time. ~24s of the 30s cron interval leaves headroom for the final batch's writes. +// +// var rather than const so tests can override it. +var reconcileWindowsProfilesScanBudget = 24 * time.Second + // ReconcileWindowsProfiles applies configuration profiles to Windows MDM hosts. -// Named return so the deferred SetCursor block below sees the actual -// function exit error. With a named return, every `return X` -// assigns X to the named err before the defer fires, so any failure -// path correctly skips the cursor write. +// +// It walks every enrolled Windows host via a host_uuid cursor (persisted in Redis through the mysqlredis wrapper), loading a +// bounded snapshot per window, computing install/remove deltas in memory (no set-difference SQL), and executing them. Within one +// tick it drains successive windows until either the delivery cap or the scan budget is hit, or the host space is exhausted +// (which resets the cursor for the next pass). +// +// Named return so the deferred SetCursor block sees the actual function-exit error: the cursor is persisted only on a clean (err +// == nil) tick, so any failure leaves the cursor untouched and the next tick re-scans from the same point. Re-scanning is cheap +// and idempotent since delivered work is now pending, so it no longer computes as work. func ReconcileWindowsProfiles(ctx context.Context, ds fleet.Datastore, logger *slog.Logger) (err error) { appConfig, err := ds.AppConfig(ctx) if err != nil { @@ -3484,89 +3501,151 @@ func ReconcileWindowsProfiles(ctx context.Context, ds fleet.Datastore, logger *s return nil } - // Read the cursor; on error, treat as start-of-pass and continue. A - // stale or missing cursor is harmless because the listing predicates - // filter out hosts whose state already matches desired state. - cursor, err := ds.GetMDMWindowsReconcileCursor(ctx) - if err != nil { - logger.WarnContext(ctx, "failed to read windows MDM reconcile cursor; starting from beginning", - "err", err) - cursor = "" + // Read the cursor; on error, treat as start-of-pass and continue. A stale or missing cursor is harmless because the in-memory + // diff installs only what actually differs from the current state. + entryCursor, cerr := ds.GetMDMWindowsReconcileCursor(ctx) + if cerr != nil { + logger.WarnContext(ctx, "failed to read windows MDM reconcile cursor; starting from beginning", "err", cerr) + entryCursor = "" } - hostUUIDs, err := ds.ListNextPendingMDMWindowsHostUUIDs(ctx, cursor, reconcileWindowsProfilesBatchSize) - if err != nil { - return ctxerr.Wrap(ctx, err, "listing next pending Windows MDM hosts") - } + cursor := entryCursor + // commitCursor is the cursor value to persist at tick end. It advances only past windows that were fully delivered; the deferred + // write fires only when err == nil, so an error leaves the cursor untouched. + commitCursor := entryCursor - if len(hostUUIDs) == 0 { - // Either no work, or we've reached the end of the cursor pass. - // Reset to "" so the next tick starts from the beginning. - // - // Decision: cursor write failures here (and in the deferred - // advance below) are logged-and-swallowed rather than returned - // as tick failures. - if cursor != "" { - logger.InfoContext(ctx, "windows MDM reconcile pass complete; resetting cursor", - "cursor", cursor) - if cerr := ds.SetMDMWindowsReconcileCursor(ctx, ""); cerr != nil { - // We assume a transient Redis failure here. - logger.WarnContext(ctx, "failed to reset windows MDM reconcile cursor", "err", cerr) - } - } - return nil - } - - // Compute the next cursor before processing so we can advance after a - // successful pass. If we got fewer than the batch size, the next tick - // should restart from the beginning. - var nextCursor string - if len(hostUUIDs) >= reconcileWindowsProfilesBatchSize { - nextCursor = hostUUIDs[len(hostUUIDs)-1] - } - - // Only log when the cursor is actually in play - i.e. the pending - // universe didn't fit in a single tick. The four state combos: - // cursor=="", nextCursor!="" - starting a multi-tick pass - // cursor!="", nextCursor!="" - continuing mid-pass - // cursor!="", nextCursor=="" - completing the final tick of a pass - // cursor=="", nextCursor=="" - silent: the entire universe fit in - // this one tick, no cursor needed - if cursor != "" || nextCursor != "" { - logger.InfoContext(ctx, "windows MDM reconcile tick using cursor", - "cursor", cursor, - "next_cursor", nextCursor, - "batch_size", reconcileWindowsProfilesBatchSize, - "hosts_in_batch", len(hostUUIDs), - ) - } - - toInstall, err := ds.ListMDMWindowsProfilesToInstallForHosts(ctx, hostUUIDs) - if err != nil { - return ctxerr.Wrap(ctx, err, "getting profiles to install") - } - toRemove, err := ds.ListMDMWindowsProfilesToRemoveForHosts(ctx, hostUUIDs) - if err != nil { - return ctxerr.Wrap(ctx, err, "getting profiles to remove") - } - - // On any error during the body below, leave the cursor where it was. - // The next tick will retry the same host window. The body's writes - // flip status from NULL to 'pending' on success, which removes those - // rows from the listing on retry, so a partial failure converges. - // - // Skip the write when the cursor isn't changing (steady-state ticks - // where the entire pending universe fit in one batch: cursor="", - // nextCursor=""). Mirrors the empty-result branch's `cursor != ""` - // guard above. defer func() { - if err == nil && cursor != nextCursor { - if cerr := ds.SetMDMWindowsReconcileCursor(ctx, nextCursor); cerr != nil { - logger.WarnContext(ctx, "failed to advance windows MDM reconcile cursor", "err", cerr) + if err == nil && commitCursor != entryCursor { + if serr := ds.SetMDMWindowsReconcileCursor(ctx, commitCursor); serr != nil { + logger.WarnContext(ctx, "failed to advance windows MDM reconcile cursor", "err", serr) } } }() + deadline := time.Now().Add(reconcileWindowsProfilesScanBudget) + deliveredHosts := 0 + + for { + hosts, allProfiles, hostLabels, currentByHost, serr := ds.GetWindowsProfileReconcileSnapshot(ctx, cursor, reconcileWindowsProfilesBatchSize) + if serr != nil { + err = ctxerr.Wrap(ctx, serr, "loading windows profile reconcile snapshot") + return err + } + + if len(hosts) == 0 { + // Reached the end of the host space (or empty fleet): reset the cursor so the next pass restarts from the beginning. + commitCursor = "" + return nil + } + + profilesByTeam := make(map[uint][]*fleet.WindowsProfileForReconcile, 4) + profilesWithBrokenLabel := make(map[string]struct{}) + for _, p := range allProfiles { + profilesByTeam[p.TeamID] = append(profilesByTeam[p.TeamID], p) + if p.HasBrokenLabel() { + profilesWithBrokenLabel[p.ProfileUUID] = struct{}{} + } + } + + toInstall, toRemove := microsoft_mdm.ComputeWindowsReconcileDeltas(hosts, hostLabels, currentByHost, profilesByTeam, profilesWithBrokenLabel) + + // Apply the per-tick delivery cap at host granularity. Hosts come back ascending by uuid, so capping keeps a contiguous prefix of + // the work-hosts and the cursor can resume at the last delivered host. + workHosts := windowsHostsWithWork(hosts, toInstall, toRemove) + advanceTo := hosts[len(hosts)-1].UUID + fullBatch := len(hosts) >= reconcileWindowsProfilesBatchSize + + partial := false + if reconcileWindowsProfilesDeliveryCap > 0 { + // Invariant: deliveredHosts < cap here. We return below as soon as it reaches the cap. So remaining >= 1. + remaining := reconcileWindowsProfilesDeliveryCap - deliveredHosts + if len(workHosts) > remaining { + allowed := make(map[string]struct{}, remaining) + for _, h := range workHosts[:remaining] { + allowed[h] = struct{}{} + } + toInstall = filterWindowsPayloadsByHost(toInstall, allowed) + toRemove = filterWindowsPayloadsByHost(toRemove, allowed) + advanceTo = workHosts[remaining-1] // resume after the last delivered host + workHosts = workHosts[:remaining] + partial = true + } + } + + if len(toInstall) > 0 || len(toRemove) > 0 { + if eerr := executeWindowsProfileReconcileBatch(ctx, ds, logger, appConfig, toInstall, toRemove); eerr != nil { + err = eerr + return err + } + } + deliveredHosts += len(workHosts) + + // Advance only after a successful execute. + commitCursor = advanceTo + cursor = advanceTo + + switch { + case partial: + // Delivery cap hit mid-window; the un-delivered remainder resumes next tick from cursor = advanceTo. + return nil + case !fullBatch: + // Short window => end of the host space; reset for the next pass. + commitCursor = "" + return nil + case reconcileWindowsProfilesDeliveryCap > 0 && deliveredHosts >= reconcileWindowsProfilesDeliveryCap: + // Delivery cap reached exactly at a window boundary. + return nil + case time.Now().After(deadline): + // Scan budget exhausted; resume next tick from cursor = advanceTo. + return nil + } + // Otherwise keep draining the next window within this tick. + } +} + +// windowsHostsWithWork returns the host UUIDs that have at least one install or remove in this window, in the order hosts are +// given (ascending by uuid). The drain loop uses this both to count delivered hosts against the cap and to pick the contiguous +// prefix to deliver when the cap is reached mid-window. +func windowsHostsWithWork(hosts []*fleet.WindowsHostReconcileInfo, toInstall, toRemove []*fleet.MDMWindowsProfilePayload) []string { + work := make(map[string]struct{}) + for _, p := range toInstall { + work[p.HostUUID] = struct{}{} + } + for _, p := range toRemove { + work[p.HostUUID] = struct{}{} + } + ordered := make([]string, 0, len(work)) + for _, h := range hosts { + if _, ok := work[h.UUID]; ok { + ordered = append(ordered, h.UUID) + } + } + return ordered +} + +// filterWindowsPayloadsByHost returns only the payloads whose HostUUID is in the allowed set, preserving order. Used to trim a +// window's deltas to the hosts that fit under the per-tick delivery cap. +func filterWindowsPayloadsByHost(payloads []*fleet.MDMWindowsProfilePayload, allowed map[string]struct{}) []*fleet.MDMWindowsProfilePayload { + out := make([]*fleet.MDMWindowsProfilePayload, 0, len(payloads)) + for _, p := range payloads { + if _, ok := allowed[p.HostUUID]; ok { + out = append(out, p) + } + } + return out +} + +// executeWindowsProfileReconcileBatch runs the post-compute reconcile pipeline against the in-memory toInstall / toRemove sets +// produced by ComputeWindowsReconcileDeltas: content fetch, deleted-profile race guard, bulk command pre-build for non-variable +// profiles, per-host variable expansion, LocURI-protected generation, host-profile upserts, and managed-certificate +// bookkeeping. This is the legacy reconciler body verbatim, now invoked once per (capped) window by the drain loop above. +func executeWindowsProfileReconcileBatch( + ctx context.Context, + ds fleet.Datastore, + logger *slog.Logger, + appConfig *fleet.AppConfig, + toInstall, toRemove []*fleet.MDMWindowsProfilePayload, +) error { // toGetContents contains the IDs of all the profiles from which we // need to retrieve contents. Since the previous query returns one row // per host, it would be too expensive to retrieve the profile contents diff --git a/server/service/microsoft_mdm_test.go b/server/service/microsoft_mdm_test.go index 41f7ab097c..2bebf45c3f 100644 --- a/server/service/microsoft_mdm_test.go +++ b/server/service/microsoft_mdm_test.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "log/slog" + "sort" "strings" "testing" "time" @@ -651,16 +652,51 @@ func setupReconcilerTest(ds *mock.Store, hostToProfile map[string]*fleet.MDMWind return nil } - // The cron's batched path picks a host window first, then calls the - // scoped listings for that window. For the mock, return all host UUIDs - // from hostToProfile so the rest of the reconciler runs against the - // same set the test wants. - ds.ListNextPendingMDMWindowsHostUUIDsFunc = func(ctx context.Context, afterHostUUID string, batchSize int) ([]string, error) { + // The cron's batched path loads a snapshot (hosts + profiles + current state) per window and computes install/remove deltas in + // memory. Return every host from hostToProfile in a single window, each paired with its profile under a unique team_id so + // ComputeWindowsReconcileDeltas maps each host to exactly its mapped profile (regardless of whether two hosts share a profile). + // Empty current state => every mapped profile is a fresh install, matching the legacy listInstall fixture. The `after != ""` + // guard keeps these single-tick tests from looping (everything is delivered in the first window). + ds.GetWindowsProfileReconcileSnapshotFunc = func(ctx context.Context, after string, batch int) ( + []*fleet.WindowsHostReconcileInfo, + []*fleet.WindowsProfileForReconcile, + map[uint]map[uint]struct{}, + map[string][]*fleet.MDMWindowsProfilePayload, + error, + ) { + if after != "" { + return nil, nil, nil, nil, nil + } + // Emit ONE profile row per unique ProfileUUID (the real snapshot is profile-scoped, not per-host). Each unique profile gets its + // own team, and every host that maps to that profile is placed in that team, so ComputeWindowsReconcileDeltas fans the single + // profile out to all its hosts, exercising shared-profile grouping the way production does. Hosts are returned ascending by UUID + // to match `ORDER BY h.uuid`. hostUUIDs := make([]string, 0, len(hostToProfile)) for hostUUID := range hostToProfile { hostUUIDs = append(hostUUIDs, hostUUID) } - return hostUUIDs, nil + sort.Strings(hostUUIDs) + teamByProfile := make(map[string]uint, len(hostToProfile)) + var hosts []*fleet.WindowsHostReconcileInfo + var profiles []*fleet.WindowsProfileForReconcile + var nextHostID uint + for _, hostUUID := range hostUUIDs { + profile := hostToProfile[hostUUID] + tid, ok := teamByProfile[profile.ProfileUUID] + if !ok { + tid = uint(len(teamByProfile) + 1) + teamByProfile[profile.ProfileUUID] = tid + profiles = append(profiles, &fleet.WindowsProfileForReconcile{ + ProfileUUID: profile.ProfileUUID, + ProfileName: profile.Name, + TeamID: tid, + }) + } + nextHostID++ + hostID, teamID := nextHostID, tid + hosts = append(hosts, &fleet.WindowsHostReconcileInfo{HostID: hostID, UUID: hostUUID, TeamID: &teamID}) + } + return hosts, profiles, nil, map[string][]*fleet.MDMWindowsProfilePayload{}, nil } listInstall := func(_ context.Context, _ ...any) ([]*fleet.MDMWindowsProfilePayload, error) { @@ -1094,8 +1130,14 @@ func TestReconcileWindowsProfilesEmptyPopulation(t *testing.T) { setCalls++ return nil } - ds.ListNextPendingMDMWindowsHostUUIDsFunc = func(ctx context.Context, after string, batchSize int) ([]string, error) { - return nil, nil + ds.GetWindowsProfileReconcileSnapshotFunc = func(ctx context.Context, after string, batch int) ( + []*fleet.WindowsHostReconcileInfo, + []*fleet.WindowsProfileForReconcile, + map[uint]map[uint]struct{}, + map[string][]*fleet.MDMWindowsProfilePayload, + error, + ) { + return nil, nil, nil, nil, nil } require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) @@ -1105,6 +1147,262 @@ func TestReconcileWindowsProfilesEmptyPopulation(t *testing.T) { } } +// setReconcileWindowsBudgets sets the three drain-loop tunables for the duration of a test and restores them on cleanup. +func setReconcileWindowsBudgets(t *testing.T, scanBatch, deliveryCap int, scanBudget time.Duration) { + t.Helper() + savedBatch := reconcileWindowsProfilesBatchSize + savedCap := reconcileWindowsProfilesDeliveryCap + savedBudget := reconcileWindowsProfilesScanBudget + t.Cleanup(func() { + reconcileWindowsProfilesBatchSize = savedBatch + reconcileWindowsProfilesDeliveryCap = savedCap + reconcileWindowsProfilesScanBudget = savedBudget + }) + reconcileWindowsProfilesBatchSize = scanBatch + reconcileWindowsProfilesDeliveryCap = deliveryCap + reconcileWindowsProfilesScanBudget = scanBudget +} + +// setKeys returns the keys of a set as a slice (order unspecified). +func setKeys(m map[string]struct{}) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} + +// windowSnapshotFunc returns a GetWindowsProfileReconcileSnapshot stub that pages allHosts (ascending) into windows of the +// requested batch size, honoring the `after` cursor, and returns the given profiles for every non-empty window. Hosts present in +// `delivered` get a matching verified install row per profile, so a delivered host no longer computes as work (modeling the real +// upsert flipping rows to verified) and a later full pass is a true no-op. If calls is non-nil it is incremented per invocation. +func windowSnapshotFunc( + allHosts []string, + profiles []*fleet.WindowsProfileForReconcile, + delivered map[string]struct{}, + calls *int, +) func(context.Context, string, int) ([]*fleet.WindowsHostReconcileInfo, []*fleet.WindowsProfileForReconcile, map[uint]map[uint]struct{}, map[string][]*fleet.MDMWindowsProfilePayload, error) { + return func(_ context.Context, after string, batch int) ( + []*fleet.WindowsHostReconcileInfo, + []*fleet.WindowsProfileForReconcile, + map[uint]map[uint]struct{}, + map[string][]*fleet.MDMWindowsProfilePayload, + error, + ) { + if calls != nil { + *calls++ + } + var hosts []*fleet.WindowsHostReconcileInfo + for i, h := range allHosts { + if h > after { + hosts = append(hosts, &fleet.WindowsHostReconcileInfo{HostID: uint(i + 1), UUID: h}) //nolint:gosec + if len(hosts) == batch { + break + } + } + } + if len(hosts) == 0 { + return nil, nil, nil, nil, nil + } + currentByHost := map[string][]*fleet.MDMWindowsProfilePayload{} + for _, h := range hosts { + if _, ok := delivered[h.UUID]; !ok { + continue + } + for _, p := range profiles { + currentByHost[h.UUID] = append(currentByHost[h.UUID], &fleet.MDMWindowsProfilePayload{ + ProfileUUID: p.ProfileUUID, + HostUUID: h.UUID, + Checksum: p.Checksum, + OperationType: fleet.MDMOperationTypeInstall, + Status: &fleet.MDMDeliveryVerified, + }) + } + } + return hosts, profiles, nil, currentByHost, nil + } +} + +// newDrainLoopTestDS wires a mock.Store for ReconcileWindowsProfiles drain-loop tests: Windows MDM enabled, a cursor backed by +// *cursor, the windowing snapshot over allHosts/profiles, and the downstream execute stubs a non-variable install needs. Enqueued +// hosts are recorded in `delivered` so a later pass is a no-op. Observe results via *cursor, the `delivered` set, and *calls. +func newDrainLoopTestDS( + ds *mock.Store, + allHosts []string, + profiles []*fleet.WindowsProfileForReconcile, + delivered map[string]struct{}, + cursor *string, + calls *int, +) { + ds.AppConfigFunc = func(ctx context.Context) (*fleet.AppConfig, error) { + cfg := &fleet.AppConfig{} + cfg.MDM.WindowsEnabledAndConfigured = true + return cfg, nil + } + ds.GetMDMWindowsReconcileCursorFunc = func(ctx context.Context) (string, error) { return *cursor, nil } + ds.SetMDMWindowsReconcileCursorFunc = func(ctx context.Context, c string) error { + *cursor = c + return nil + } + ds.GetWindowsProfileReconcileSnapshotFunc = windowSnapshotFunc(allHosts, profiles, delivered, calls) + ds.GetMDMWindowsProfilesContentsFunc = func(ctx context.Context, uuids []string) (map[string]fleet.MDMWindowsProfileContents, error) { + out := map[string]fleet.MDMWindowsProfileContents{} + for _, p := range profiles { + out[p.ProfileUUID] = fleet.MDMWindowsProfileContents{ + SyncML: []byte(`./Testv`), + Checksum: p.Checksum, + } + } + return out, nil + } + ds.GetExistingMDMWindowsProfileUUIDsFunc = func(ctx context.Context, uuids []string) (map[string]struct{}, error) { + out := map[string]struct{}{} + for _, p := range profiles { + out[p.ProfileUUID] = struct{}{} + } + return out, nil + } + ds.GetGroupedCertificateAuthoritiesFunc = func(ctx context.Context, includeSecrets bool) (*fleet.GroupedCertificateAuthorities, error) { + return &fleet.GroupedCertificateAuthorities{}, nil + } + ds.MDMWindowsBulkInsertCommandsFunc = func(ctx context.Context, cmds []*fleet.MDMWindowsCommand) error { return nil } + ds.MDMWindowsEnqueueCommandAndUpsertHostProfilesFunc = func(ctx context.Context, hostUUIDs []string, cmd *fleet.MDMWindowsCommand, payload []*fleet.MDMWindowsBulkUpsertHostProfilePayload) error { + for _, h := range hostUUIDs { + delivered[h] = struct{}{} + } + return nil + } + ds.BulkUpsertMDMWindowsHostProfilesFunc = func(ctx context.Context, payload []*fleet.MDMWindowsBulkUpsertHostProfilePayload) error { + return nil + } + ds.BulkUpsertMDMManagedCertificatesFunc = func(ctx context.Context, payload []*fleet.MDMManagedCertificate) error { + return nil + } +} + +// TestReconcileWindowsProfilesDeliveryCapThrottlesPerTick exercises the within-tick drain loop's delivery cap: with a large scan +// window but a small per-tick delivery cap, a bulk change (every enrolled host needs the same profile) is throttled to +// deliveryCap hosts per tick, the cursor advances only to the last delivered host, and successive ticks drain the remainder until +// the host space is exhausted (cursor resets to ""). This preserves the writer-pressure smoothing the legacy 2000-host batch +// provided. +func TestReconcileWindowsProfilesDeliveryCapThrottlesPerTick(t *testing.T) { + ctx := context.Background() + ds := new(mock.Store) + logger := slog.New(slog.DiscardHandler) + + // Large scan window (the whole fleet fits in one window), small delivery cap, no wall-clock limit. + setReconcileWindowsBudgets(t, 100 /*scanBatch*/, 3 /*deliveryCap*/, time.Hour) + + allHosts := []string{"h00", "h01", "h02", "h03", "h04", "h05", "h06", "h07", "h08", "h09"} + profiles := []*fleet.WindowsProfileForReconcile{{ProfileUUID: "shared-profile", ProfileName: "Shared", TeamID: 0, Checksum: []byte("c")}} + delivered := map[string]struct{}{} + var cursor string + newDrainLoopTestDS(ds, allHosts, profiles, delivered, &cursor, nil) + + // Capture exactly which hosts each enqueue delivered (still marking them delivered for convergence). + var deliveredBatches [][]string + ds.MDMWindowsEnqueueCommandAndUpsertHostProfilesFunc = func(ctx context.Context, hostUUIDs []string, cmd *fleet.MDMWindowsCommand, payload []*fleet.MDMWindowsBulkUpsertHostProfilePayload) error { + deliveredBatches = append(deliveredBatches, append([]string{}, hostUUIDs...)) + for _, h := range hostUUIDs { + delivered[h] = struct{}{} + } + return nil + } + + // Tick 1: deliver the first 3 hosts (contiguous prefix); cursor advances to the last delivered host. + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Equal(t, "h02", cursor) + require.Equal(t, [][]string{{"h00", "h01", "h02"}}, deliveredBatches) + + // Ticks 2-3: next 3 hosts each. + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Equal(t, "h05", cursor) + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Equal(t, "h08", cursor) + + // Tick 4: final host (short window) drains and resets the cursor. + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Empty(t, cursor) + + // Tick 5: empty fleet pass, cursor stays reset, nothing re-delivered. + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Empty(t, cursor) + + // Every host was delivered exactly once across the ticks. + var all []string + for _, b := range deliveredBatches { + all = append(all, b...) + } + require.ElementsMatch(t, allHosts, all) +} + +// TestReconcileWindowsProfilesDrainsMultipleWindowsPerTick covers the core drain behavior the other tests don't: when the delivery +// cap spans several scan windows, one tick reads window after window (cheap indexed reads) accumulating delivered hosts until the +// cap is reached mid-window. With scanBatch=2 and cap=5 over 6 hosts that all need work, tick 1 makes 3 snapshot reads (delivering +// 2+2+1) and stops at the 5th host; tick 2 delivers the remainder and resets the cursor. +func TestReconcileWindowsProfilesDrainsMultipleWindowsPerTick(t *testing.T) { + ctx := context.Background() + ds := new(mock.Store) + logger := slog.New(slog.DiscardHandler) + + setReconcileWindowsBudgets(t, 2 /*scanBatch*/, 5 /*deliveryCap*/, time.Hour) + + allHosts := []string{"h0", "h1", "h2", "h3", "h4", "h5"} + profiles := []*fleet.WindowsProfileForReconcile{{ProfileUUID: "p", ProfileName: "P", TeamID: 0, Checksum: []byte("c")}} + delivered := map[string]struct{}{} + var cursor string + var snapshotCalls int + newDrainLoopTestDS(ds, allHosts, profiles, delivered, &cursor, &snapshotCalls) + + // Tick 1: drains 3 windows (2+2+1) to reach the cap of 5, stopping mid-third-window at h4. + snapshotCalls = 0 + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Equal(t, 3, snapshotCalls, "one tick should read multiple windows to fill the cap") + require.Equal(t, "h4", cursor) + require.ElementsMatch(t, []string{"h0", "h1", "h2", "h3", "h4"}, setKeys(delivered)) + + // Tick 2: delivers the last host; the short final window resets the cursor. + snapshotCalls = 0 + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Empty(t, cursor) + require.ElementsMatch(t, allHosts, setKeys(delivered)) + + // Tick 3: full no-op pass over the now all-delivered fleet, cursor stays reset. + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Empty(t, cursor) +} + +// TestReconcileWindowsProfilesScanBudgetHaltsDrain exercises the scan-budget branch of the drain loop: when the wall-clock budget +// is already exhausted, the tick stops after the first scanned window and persists the cursor at the last scanned host (it does +// NOT keep draining to the end of the fleet, and does NOT reset the cursor). The next tick resumes from there. +func TestReconcileWindowsProfilesScanBudgetHaltsDrain(t *testing.T) { + ctx := context.Background() + ds := new(mock.Store) + logger := slog.New(slog.DiscardHandler) + + // Small windows, generous delivery cap (so the cap never governs), and an already-expired scan budget so the loop halts after + // the first window. + setReconcileWindowsBudgets(t, 2 /*scanBatch*/, 1000 /*deliveryCap*/, time.Nanosecond) + + allHosts := []string{"h0", "h1", "h2", "h3", "h4", "h5"} + // No profiles => no work; this test is purely about the scan/cursor mechanics, so execute is never reached. + delivered := map[string]struct{}{} + var cursor string + var snapshotCalls int + newDrainLoopTestDS(ds, allHosts, nil /*profiles*/, delivered, &cursor, &snapshotCalls) + + // Tick 1: the budget is already spent, so only the first window is scanned and the cursor advances to its last host (not reset). + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Equal(t, 1, snapshotCalls) + require.Equal(t, "h1", cursor) + + // Tick 2: resumes from the persisted cursor, reads the NEXT window and advances again, confirming progress isn't lost. + snapshotCalls = 0 + require.NoError(t, ReconcileWindowsProfiles(ctx, ds, logger)) + require.Equal(t, 1, snapshotCalls) + require.Equal(t, "h3", cursor) +} + func TestRekeyWindowsDevice(t *testing.T) { ds := new(mock.Store) kv := new(mock.KVStore) diff --git a/server/service/reconcile_windows_profiles_property_test.go b/server/service/reconcile_windows_profiles_property_test.go index eb3b880faf..72cbc5639e 100644 --- a/server/service/reconcile_windows_profiles_property_test.go +++ b/server/service/reconcile_windows_profiles_property_test.go @@ -6,6 +6,7 @@ import ( "log/slog" "slices" "testing" + "time" "github.com/fleetdm/fleet/v4/server/fleet" "github.com/fleetdm/fleet/v4/server/mock" @@ -13,30 +14,38 @@ import ( "pgregory.net/rapid" ) -// Property-based tests for the cursor state machine in -// ReconcileWindowsProfiles. These cover invariants that emerge across many -// cron ticks (coverage, monotonicity, failure semantics) which the existing -// table-driven tests cannot exercise directly. +// Property-based tests for the cursor state machine in ReconcileWindowsProfiles. These cover invariants that emerge across many +// cron ticks (coverage, monotonicity, failure semantics) which the existing table-driven tests cannot exercise directly. +// +// The reconciler walks every enrolled Windows host via GetWindowsProfileReconcileSnapshot and drains successive windows within a +// tick until a budget is hit. To pin the cursor protocol to a deterministic one-window-per-tick cadence, these tests set the +// delivery cap equal to the scan batch and make every host have exactly one pending install (one global, label-less profile + +// empty current state). The scan budget is set large so only the delivery cap governs. // // Run with more checks: // go test -run TestPBT_ReconcileWindowsProfiles ./server/service/ -args -rapid.checks=2000 -// cursorFakeState is the in-memory model the fake datastore exposes to the -// cron. It mirrors only what ReconcileWindowsProfiles depends on: a sorted -// set of host UUIDs with pending work, the persisted Redis-style cursor, -// and a per-host visit counter so tests can assert coverage. +// cursorFakeState is the in-memory model the fake datastore exposes to the cron. It mirrors only what ReconcileWindowsProfiles +// depends on: a sorted set of enrolled host UUIDs, the persisted Redis-style cursor, and a per-host visit counter so tests can +// assert coverage. type cursorFakeState struct { cursor string - pending []string // sorted ascending; mirrors what ListNextPendingMDMWindowsHostUUIDs returns + pending []string // sorted ascending; the enrolled Windows host universe the snapshot pages through visited map[string]int cursorSet int // number of times SetMDMWindowsReconcileCursor was called } -// newCursorFakeDS wires a mock.Store with funcs that route to cursorFakeState -// and stubs every other DS method ReconcileWindowsProfiles calls so the body -// is a successful no-op end-to-end (no install/remove targets, no upserts). -// That keeps the property tests focused on the cursor protocol; the body's -// per-profile behavior is covered by the existing table-driven tests. +// newCursorFakeDS wires a mock.Store with funcs that route to cursorFakeState and stubs every other DS method +// ReconcileWindowsProfiles calls so the body runs end-to-end without enqueueing real work. The snapshot returns the windowed +// hosts plus a single global profile so every host computes as one install; the existence pre-check then reports the profile +// gone, so execute short-circuits without commands. That keeps the property tests focused on the cursor protocol while still +// driving the delivery-cap accounting. +// +// COUPLING NOTE: the fake never actually enqueues, so the delivery cap is reached only because ReconcileWindowsProfiles counts +// intended work (pre-execute workHosts), not actual deliveries. If that accounting is ever changed to count only +// actually-scheduled hosts (the deferred CodeRabbit/Copilot review point), this fake would report zero delivered, the cap would +// never be reached, and the one-window-per-tick assumption these tests rely on (plus the coverage/monotonic/advance assertions) +// would break. The fake would then need to actually deliver work to keep driving the cap. func newCursorFakeDS(initialHosts []string) (*mock.Store, *cursorFakeState) { sorted := slices.Clone(initialHosts) slices.Sort(sorted) @@ -56,87 +65,100 @@ func newCursorFakeDS(initialHosts []string) (*mock.Store, *cursorFakeState) { state.cursorSet++ return nil } - ds.ListNextPendingMDMWindowsHostUUIDsFunc = func(ctx context.Context, after string, batch int) ([]string, error) { - var out []string - for _, h := range state.pending { + ds.GetWindowsProfileReconcileSnapshotFunc = func(ctx context.Context, after string, batch int) ( + []*fleet.WindowsHostReconcileInfo, + []*fleet.WindowsProfileForReconcile, + map[uint]map[uint]struct{}, + map[string][]*fleet.MDMWindowsProfilePayload, + error, + ) { + var window []*fleet.WindowsHostReconcileInfo + for i, h := range state.pending { if h > after { - out = append(out, h) - if len(out) == batch { + window = append(window, &fleet.WindowsHostReconcileInfo{HostID: uint(i + 1), UUID: h}) //nolint:gosec + if len(window) == batch { break } } } - for _, h := range out { - state.visited[h]++ + for _, h := range window { + state.visited[h.UUID]++ } - return out, nil - } - // Empty install/remove/contents/existence: the cron's body sails - // through without enqueueing any work, the deferred SetCursor still - // fires, and the cursor protocol is exercised in isolation. - ds.ListMDMWindowsProfilesToInstallForHostsFunc = func(ctx context.Context, hostUUIDs []string) ([]*fleet.MDMWindowsProfilePayload, error) { - return nil, nil - } - ds.ListMDMWindowsProfilesToRemoveForHostsFunc = func(ctx context.Context, hostUUIDs []string) ([]*fleet.MDMWindowsProfilePayload, error) { - return nil, nil + if len(window) == 0 { + return nil, nil, nil, nil, nil + } + // One global, label-less profile so every host in the window computes as exactly one install (current state is empty), so + // each host counts once against the per-tick delivery cap. + profiles := []*fleet.WindowsProfileForReconcile{ + {ProfileUUID: "p-global", ProfileName: "Global", TeamID: 0, Checksum: []byte("c")}, + } + return window, profiles, nil, map[string][]*fleet.MDMWindowsProfilePayload{}, nil } ds.GetMDMWindowsProfilesContentsFunc = func(ctx context.Context, uuids []string) (map[string]fleet.MDMWindowsProfileContents, error) { return map[string]fleet.MDMWindowsProfileContents{}, nil } + // Report the profile as gone so the install loop short-circuits without enqueueing commands; the cursor protocol is unaffected. ds.GetExistingMDMWindowsProfileUUIDsFunc = func(ctx context.Context, uuids []string) (map[string]struct{}, error) { return map[string]struct{}{}, nil } - // The body unconditionally calls these two upserts at the end of every - // tick (even with an empty payload). Stub them as no-ops. + // The body unconditionally calls these two upserts at the end of every window it executes (even with an empty payload). Stub them + // as no-ops. ds.BulkUpsertMDMWindowsHostProfilesFunc = func(ctx context.Context, payload []*fleet.MDMWindowsBulkUpsertHostProfilePayload) error { return nil } ds.BulkUpsertMDMManagedCertificatesFunc = func(ctx context.Context, payload []*fleet.MDMManagedCertificate) error { return nil } - // GetGroupedCertificateAuthorities is called unconditionally on every - // tick (even when there is no profile content to process). Return a - // zero-value struct so the dependent maps are empty and the rest of - // the body short-circuits. + // GetGroupedCertificateAuthorities is called whenever a window has work. Return a zero-value struct so the dependent maps are + // empty and the rest of the body short-circuits. ds.GetGroupedCertificateAuthoritiesFunc = func(ctx context.Context, includeSecrets bool) (*fleet.GroupedCertificateAuthorities, error) { return &fleet.GroupedCertificateAuthorities{}, nil } return ds, state } -// pbtBatchSizeOverride installs a property-test-scoped batch size on the -// package-level reconcileWindowsProfilesBatchSize var. The single -// t.Cleanup restores the original value once the whole test (across all -// rapid trials) finishes; per-trial reassignments inside the rapid.Check -// closure overwrite each other, which is the desired behavior. -func pbtBatchSizeOverride(t *testing.T) { +// pbtBudgetOverride installs property-test-scoped budgets on the package-level reconciler tunables. The single t.Cleanup restores +// the original values once the whole test (across all rapid trials) finishes; per-trial reassignments inside the rapid.Check +// closure overwrite each other, which is the desired behavior. Tests set the delivery cap equal to the scan batch (one window per +// tick) and leave the scan budget large so only the cap governs. +func pbtBudgetOverride(t *testing.T) { t.Helper() - saved := reconcileWindowsProfilesBatchSize - t.Cleanup(func() { reconcileWindowsProfilesBatchSize = saved }) + savedBatch := reconcileWindowsProfilesBatchSize + savedCap := reconcileWindowsProfilesDeliveryCap + savedBudget := reconcileWindowsProfilesScanBudget + t.Cleanup(func() { + reconcileWindowsProfilesBatchSize = savedBatch + reconcileWindowsProfilesDeliveryCap = savedCap + reconcileWindowsProfilesScanBudget = savedBudget + }) } -// hostGen draws short distinct strings; only their relative lexicographic -// order matters to the cursor protocol, not that they look like real UUIDs. +// pbtSetBudgets pins the per-trial budgets: scan batch == delivery cap (one window of delivered work per tick) and a large scan +// budget so wall clock never ends a tick early. +func pbtSetBudgets(batch int) { + reconcileWindowsProfilesBatchSize = batch + reconcileWindowsProfilesDeliveryCap = batch + reconcileWindowsProfilesScanBudget = time.Hour +} + +// hostGen draws short distinct strings; only their relative lexicographic order matters to the cursor protocol. var hostGen = rapid.StringMatching(`[a-z]{1,8}`) // pbtLogger discards everything var pbtLogger = slog.New(slog.DiscardHandler) -// TestPBT_ReconcileWindowsProfilesCoverage verifies that for any stable -// population of pending hosts, the cursor protocol reaches a state where -// every host has been visited and the cursor has returned to "" within a -// bounded number of ticks. The bound is ⌈N/B⌉+2 to absorb the extra -// "empty pass after exact-multiple full pass" tick. +// TestPBT_ReconcileWindowsProfilesCoverage verifies that for any stable population of enrolled hosts, the cursor protocol reaches +// a state where every host has been visited and the cursor has returned to "" within a bounded number of ticks. The bound is +// ⌈N/B⌉+2 to absorb the extra "empty pass after exact-multiple full pass" tick. // -// We stop ticking as soon as that joint state is reached because any -// further tick restarts the pass (cursor moves back off ""), which would -// make a fixed-tick-count assertion brittle. +// We stop ticking as soon as that joint state is reached because any further tick restarts the pass (cursor moves back off ""), +// which would make a fixed-tick-count assertion brittle. func TestPBT_ReconcileWindowsProfilesCoverage(t *testing.T) { - pbtBatchSizeOverride(t) + pbtBudgetOverride(t) rapid.Check(t, func(rt *rapid.T) { batch := rapid.IntRange(1, 25).Draw(rt, "batchSize") hosts := rapid.SliceOfNDistinct(hostGen, 0, 200, rapid.ID[string]).Draw(rt, "hosts") - reconcileWindowsProfilesBatchSize = batch + pbtSetBudgets(batch) ds, state := newCursorFakeDS(hosts) ctx := t.Context() @@ -169,16 +191,15 @@ func TestPBT_ReconcileWindowsProfilesCoverage(t *testing.T) { }) } -// TestPBT_ReconcileWindowsProfilesMonotonic verifies that within a pass the -// cursor strictly increases between two consecutive non-reset ticks. Reset -// transitions ("" -> non-empty starting fresh, or non-empty -> "" at end of -// pass) are allowed and expected. +// TestPBT_ReconcileWindowsProfilesMonotonic verifies that within a pass the cursor strictly increases between two consecutive +// non-reset ticks. Reset transitions ("" -> non-empty starting fresh, or non-empty -> "" at end of pass) are allowed and +// expected. func TestPBT_ReconcileWindowsProfilesMonotonic(t *testing.T) { - pbtBatchSizeOverride(t) + pbtBudgetOverride(t) rapid.Check(t, func(rt *rapid.T) { batch := rapid.IntRange(1, 20).Draw(rt, "batchSize") hosts := rapid.SliceOfNDistinct(hostGen, 1, 100, rapid.ID[string]).Draw(rt, "hosts") - reconcileWindowsProfilesBatchSize = batch + pbtSetBudgets(batch) ds, state := newCursorFakeDS(hosts) ctx := t.Context() @@ -198,60 +219,42 @@ func TestPBT_ReconcileWindowsProfilesMonotonic(t *testing.T) { }) } -// TestPBT_ReconcileWindowsProfilesFailureNoAdvance verifies the universal -// invariant "any body failure leaves the cursor untouched." The cron's -// SetCursor write is gated by a named-return-aware defer that skips on -// error, plus pre-defer failure paths that simply return without -// registering the defer at all. The property must hold for both classes, -// and rapid randomly samples across them so a regression in either path -// surfaces here. -// -// Catches regressions like: -// - Switching the named-return-aware defer to a bare `return ...` -// mid-body (would advance on post-defer failure). -// - Moving the defer registration earlier so a pre-listing failure no -// longer skips it (would advance with a stale nextCursor). -// - Adding a SetCursor call elsewhere in the body that fires before -// the err check. +// TestPBT_ReconcileWindowsProfilesFailureNoAdvance verifies the universal invariant "any body failure leaves the cursor +// untouched." The cron's SetCursor write is gated by a named-return-aware defer that skips on error, and commitCursor only +// advances past a fully-delivered window, so a failure in the first window never persists a cursor. rapid randomly samples across +// pre-execute and in-execute failure points so a regression in either path surfaces here. func TestPBT_ReconcileWindowsProfilesFailureNoAdvance(t *testing.T) { - pbtBatchSizeOverride(t) + pbtBudgetOverride(t) rapid.Check(t, func(rt *rapid.T) { batch := rapid.IntRange(1, 10).Draw(rt, "batchSize") - // Need at least one host so the host listing returns non-empty - // when it succeeds; otherwise the cron takes the empty-pop early - // return and does not exercise the failure path we want. + // Need at least one host so the snapshot returns a non-empty window when it succeeds; otherwise the cron takes the empty-pop + // early return and does not exercise the failure path we want. hosts := rapid.SliceOfNDistinct(hostGen, 1, 30, rapid.ID[string]).Draw(rt, "hosts") failurePoint := rapid.SampledFrom([]string{ - "ListNextPendingMDMWindowsHostUUIDs", // pre-defer - "ListMDMWindowsProfilesToInstallForHosts", // pre-defer - "ListMDMWindowsProfilesToRemoveForHosts", // pre-defer - "GetMDMWindowsProfilesContents", // post-defer - "GetGroupedCertificateAuthorities", // post-defer - "BulkUpsertMDMWindowsHostProfiles", // post-defer (end of body) + "GetWindowsProfileReconcileSnapshot", // pre-execute + "GetMDMWindowsProfilesContents", // in execute + "GetGroupedCertificateAuthorities", // in execute + "BulkUpsertMDMWindowsHostProfiles", // in execute (end of body) }).Draw(rt, "failurePoint") - reconcileWindowsProfilesBatchSize = batch + pbtSetBudgets(batch) ds, state := newCursorFakeDS(hosts) - // Seed a non-empty cursor so "cursor untouched on failure" is a - // real assertion rather than trivially-true on the empty default. - // "0" sorts before any value hostGen can produce ([a-z]{1,8}), so - // the host listing still returns every host and the cron reaches - // the injected failure point. + // Seed a non-empty cursor so "cursor untouched on failure" is a real assertion rather than trivially-true on the empty default. + // "0" sorts before any value hostGen can produce ([a-z]{1,8}), so the snapshot still returns the first window and the cron + // reaches the injected failure point. const initialCursor = "0" state.cursor = initialCursor simErr := errors.New("simulated failure at " + failurePoint) switch failurePoint { - case "ListNextPendingMDMWindowsHostUUIDs": - ds.ListNextPendingMDMWindowsHostUUIDsFunc = func(ctx context.Context, after string, b int) ([]string, error) { - return nil, simErr - } - case "ListMDMWindowsProfilesToInstallForHosts": - ds.ListMDMWindowsProfilesToInstallForHostsFunc = func(ctx context.Context, hostUUIDs []string) ([]*fleet.MDMWindowsProfilePayload, error) { - return nil, simErr - } - case "ListMDMWindowsProfilesToRemoveForHosts": - ds.ListMDMWindowsProfilesToRemoveForHostsFunc = func(ctx context.Context, hostUUIDs []string) ([]*fleet.MDMWindowsProfilePayload, error) { - return nil, simErr + case "GetWindowsProfileReconcileSnapshot": + ds.GetWindowsProfileReconcileSnapshotFunc = func(ctx context.Context, after string, batch int) ( + []*fleet.WindowsHostReconcileInfo, + []*fleet.WindowsProfileForReconcile, + map[uint]map[uint]struct{}, + map[string][]*fleet.MDMWindowsProfilePayload, + error, + ) { + return nil, nil, nil, nil, simErr } case "GetMDMWindowsProfilesContents": ds.GetMDMWindowsProfilesContentsFunc = func(ctx context.Context, uuids []string) (map[string]fleet.MDMWindowsProfileContents, error) { @@ -277,39 +280,85 @@ func TestPBT_ReconcileWindowsProfilesFailureNoAdvance(t *testing.T) { }) } -// TestPBT_ReconcileWindowsProfilesCursorAdvanceMatchesLastVisited verifies -// the per-tick cursor invariant: after a non-empty tick, either the cursor -// equals the lexicographically last host visited in that tick (full batch), -// or it is "" (short batch, signaling end of pass). +// TestPBT_ReconcileWindowsProfilesFailureNoAdvanceMultiWindow extends the no-advance-on-error invariant into the multi-window +// drain regime: with the delivery cap set high so one tick drains several windows, a failure on a LATER window (after earlier +// windows in the same tick already succeeded) must still leave the cursor untouched. This guards against a regression that +// persists per-window progress mid-tick (e.g. moving SetCursor inside the loop), which the single-window FailureNoAdvance test +// above cannot catch because there the failing window is always the first. +func TestPBT_ReconcileWindowsProfilesFailureNoAdvanceMultiWindow(t *testing.T) { + pbtBudgetOverride(t) + rapid.Check(t, func(rt *rapid.T) { + batch := rapid.IntRange(1, 5).Draw(rt, "batchSize") + // At least 6 hosts (> max batch) guarantees the tick spans >= 2 windows. + hosts := rapid.SliceOfNDistinct(hostGen, 6, 40, rapid.ID[string]).Draw(rt, "hosts") + numWindows := (len(hosts) + batch - 1) / batch + failWindow := rapid.IntRange(2, numWindows).Draw(rt, "failWindow") + + // Large cap and scan budget so neither ends the tick early; only the injected failure stops it, after failWindow-1 successful + // windows. + reconcileWindowsProfilesBatchSize = batch + reconcileWindowsProfilesDeliveryCap = 1_000_000 + reconcileWindowsProfilesScanBudget = time.Hour + + ds, state := newCursorFakeDS(hosts) + const initialCursor = "0" + state.cursor = initialCursor + + // Every window has work, so GetMDMWindowsProfilesContents is called once per window. Fail the failWindow-th call; earlier windows + // succeed. + simErr := errors.New("simulated failure") + contentsCalls := 0 + ds.GetMDMWindowsProfilesContentsFunc = func(ctx context.Context, uuids []string) (map[string]fleet.MDMWindowsProfileContents, error) { + contentsCalls++ + if contentsCalls == failWindow { + return nil, simErr + } + return map[string]fleet.MDMWindowsProfileContents{}, nil + } + + err := ReconcileWindowsProfiles(t.Context(), ds, pbtLogger) + require.Errorf(rt, err, "failure on window %d/%d did not propagate", failWindow, numWindows) + // Confirms earlier windows really ran (so the precondition isn't vacuous). + require.Equalf(rt, failWindow, contentsCalls, + "expected to reach window %d before failing (N=%d, B=%d)", failWindow, len(hosts), batch) + require.Equalf(rt, initialCursor, state.cursor, + "cursor advanced despite failure on window %d after %d successful windows", failWindow, failWindow-1) + require.Equalf(rt, 0, state.cursorSet, + "SetMDMWindowsReconcileCursor called despite mid-tick failure on window %d", failWindow) + }) +} + +// TestPBT_ReconcileWindowsProfilesCursorAdvanceMatchesLastVisited verifies the per-tick cursor invariant: with delivery cap == +// scan batch and every host having work, each tick delivers exactly one window, so after a non-empty tick the cursor equals the +// lexicographically last host in that window (full batch), or it is "" (short batch, signaling end of pass). func TestPBT_ReconcileWindowsProfilesCursorAdvanceMatchesLastVisited(t *testing.T) { - pbtBatchSizeOverride(t) + pbtBudgetOverride(t) rapid.Check(t, func(rt *rapid.T) { batch := rapid.IntRange(1, 20).Draw(rt, "batchSize") hosts := rapid.SliceOfNDistinct(hostGen, 1, 100, rapid.ID[string]).Draw(rt, "hosts") - reconcileWindowsProfilesBatchSize = batch + pbtSetBudgets(batch) sorted := slices.Clone(hosts) slices.Sort(sorted) ds, state := newCursorFakeDS(hosts) ctx := t.Context() - // Walk the population by ticks; at each step, predict the batch - // from sorted/cursor and check the resulting cursor matches the + // Walk the population by ticks; at each step, predict the window from sorted/cursor and check the resulting cursor matches the // rule. seen := 0 for seen < len(sorted) { - expectedBatch := sorted[seen:min(seen+batch, len(sorted))] + expectedWindow := sorted[seen:min(seen+batch, len(sorted))] require.NoError(rt, ReconcileWindowsProfiles(ctx, ds, pbtLogger)) - if len(expectedBatch) >= batch { - require.Equalf(rt, expectedBatch[len(expectedBatch)-1], state.cursor, - "full batch must leave cursor at last UUID; expected=%q got=%q", - expectedBatch[len(expectedBatch)-1], state.cursor) + if len(expectedWindow) >= batch { + require.Equalf(rt, expectedWindow[len(expectedWindow)-1], state.cursor, + "full window must leave cursor at last UUID; expected=%q got=%q", + expectedWindow[len(expectedWindow)-1], state.cursor) } else { require.Emptyf(rt, state.cursor, - "short batch must leave cursor empty; got=%q", state.cursor) + "short window must leave cursor empty; got=%q", state.cursor) } - seen += len(expectedBatch) + seen += len(expectedWindow) } }) }