Improved Windows MDM reconciler (#47071)
<!-- Add the related story/sub-task/bug number, like Resolves #123, or remove if NA --> **Related issue:** Resolves #45635 Moved profile reconciler work from SQL to code, similar to what Apple MDM team did last sprint. The Windows MDM loadtest for 40 profiles with 30K hosts looks much better. ``` ┌──────────────────────────┬─────────────────────────────────────────┬───────────────────────────────────────────┐ │ │ Pre-fix baseline │ This branch │ ├──────────────────────────┼─────────────────────────────────────────┼───────────────────────────────────────────┤ │ Transfer wall time │ ~40–42 min │ ~15.5 min │ ├──────────────────────────┼─────────────────────────────────────────┼───────────────────────────────────────────┤ │ Per work tick │ 215–257s (host-finding query dominated) │ ~48s (host-finding gone; now bulk writes) │ ├──────────────────────────┼─────────────────────────────────────────┼───────────────────────────────────────────┤ │ Ticks > 30s (work ticks) │ ~all │ ~all (16/17, ~48s) │ ├──────────────────────────┼─────────────────────────────────────────┼───────────────────────────────────────────┤ │ Pacing governor │ the host-finding query │ the 2000-host delivery cap + 30s interval │ └──────────────────────────┴─────────────────────────────────────────┴───────────────────────────────────────────┘ ``` The writer spikes briefly to 16 AAS, but has CPU headroom, so I'd say we can claim to support 40 profies on 30K hosts. ``` ┌───────────────┬───────────────┬───────────────────────────┬───────────────────────────────────────┐ │ Instance │ CPU avg / max │ DBLoad avg / max (4 vCPU) │ Read / Write / Commit latency │ ├───────────────┼───────────────┼───────────────────────────┼───────────────────────────────────────┤ │ writer -two │ 67.9% / 74.9% │ 4.84 / 16.0 │ 0.28ms / 2.07ms / 10.6ms (max 12.2ms) │ ├───────────────┼───────────────┼───────────────────────────┼───────────────────────────────────────┤ │ reader -one │ 46.1% / 52.7% │ 1.19 / 5.0 │ 1.59ms / — / — │ ├───────────────┼───────────────┼───────────────────────────┼───────────────────────────────────────┤ │ reader -three │ 65.4% / 70.1% │ 1.77 / 5.0 │ 1.72ms / — / — │ └───────────────┴───────────────┴───────────────────────────┴───────────────────────────────────────┘ ``` # Checklist for submitter If some of the following don't apply, delete the relevant line. - [x] Changes file added for user-visible changes in `changes/`, `orbit/changes/` or `ee/fleetd-chrome/changes`. See [Changes files](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/guides/committing-changes.md#changes-files) for more information. ## Testing - [x] Added/updated automated tests - [x] Where appropriate, [automated tests simulate multiple hosts and test for host isolation](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/reference/patterns-backend.md#unit-testing) (updates to one hosts's records do not affect another) - [x] QA'd all new/changed functionality manually <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Refactor** * Reworked Windows MDM reconciliation to a snapshot-based, batched drain-loop, improving responsiveness and reducing database load during large profile operations. * **Performance / Reliability** * Windows MDM profile changes now reach hosts faster; large team-wide profile additions/removals (including host transfers) complete more quickly with lower DB impact. * **Chore / Configuration** * Added tunables to control per-tick delivery caps and scan budgets. * **Tests** * Expanded end-to-end and property tests for install/remove, team/label gating, and multi-window drain behavior. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Victor Lyuboslavsky
@@ -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.
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"))
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+165
-86
@@ -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 <Delete> 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
|
||||
|
||||
@@ -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(`<Replace><Item><Target><LocURI>./Test</LocURI></Target><Data>v</Data></Item></Replace>`),
|
||||
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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user