package console

// The on-demand scan drain under host claims (scan-revival claims design).
//
// These drive objstore.DrainApp, RebuildQueue and CollectScanResults through
// the enabled console's own boundary store and guarded scanner, and read the
// canonical admission afterwards. What they pin is the operation SHAPE, which
// no wrapper call count can: one short exclusive operation per scan group, a
// group that completes is durable on its own, a failure before any effect
// reopens the app, an uncertain Scan holds exactly the operation it ran under,
// known outcomes are results and not failures, housekeeping opens an
// operation only when there is something to delete, and nothing is deleted or
// scanned while another operation owns the app.

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/objstore"
)

// scanClaimWorld is one enabled console with bab/titogo granted for scanning.
// Items are staged through the RAW bucket at chosen times (world staging, not
// a door under test); the drain runs through the boundary.
type scanClaimWorld struct {
	cx     *claimCase
	bucket string
	appID  string
	ref    claimAppRef
}

func newScanClaimWorld(t *testing.T) *scanClaimWorld {
	t.Helper()
	cx := newClaimCase(t)
	cx.scanGranted(t)
	rec, _, err := objstore.Get(context.Background(), cx.rel, "bab", "titogo")
	if err != nil || rec == nil || rec.Provisioned == nil {
		t.Fatalf("granted store record %+v %v", rec, err)
	}
	ref := claimAppRef{Account: "bab", App: "titogo"}
	return &scanClaimWorld{cx: cx, bucket: rec.Provisioned.Bucket, appID: cx.appIDs[ref], ref: ref}
}

// seed puts key in the fake engine's view of the bucket and returns its version.
func (w *scanClaimWorld) seed(t *testing.T, key string) string {
	t.Helper()
	w.cx.scanFake.PutObject(w.bucket, key, `"etag-`+key+`"`, time.Now().Add(-time.Hour), 10)
	_, obj, err := w.cx.scanFake.Status(context.Background(), w.bucket, key)
	if err != nil {
		t.Fatal(err)
	}
	return obj.Version()
}

// enqueue stages one item at a chosen time through the raw bucket, so groups
// order the way the test says rather than by the millisecond.
func (w *scanClaimWorld) enqueue(t *testing.T, key string, at time.Time) {
	t.Helper()
	if _, err := objstore.Enqueue(context.Background(), w.cx.rel, w.cx.srv.scanQueue, "bab", "titogo",
		&objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
		t.Fatalf("stage queue item %s: %v", key, err)
	}
}

func (w *scanClaimWorld) drain(t *testing.T) ([]objstore.Drained, error) {
	t.Helper()
	done, _, err := objstore.DrainApp(context.Background(), w.cx.srv.cfg.Releases, w.cx.srv.cfg.StoreScanner,
		w.cx.srv.scanQueue, "bab", "titogo", scanDrainPerApp, time.Now())
	return done, err
}

func (w *scanClaimWorld) admission(t *testing.T) hostops.AppAdmission {
	t.Helper()
	a, err := hostops.ReadCanonicalAdmission(context.Background(), w.cx.rel, "bab", "titogo")
	if err != nil {
		t.Fatalf("read admission: %v", err)
	}
	return a.Value
}

func (w *scanClaimWorld) queued(t *testing.T) []string {
	t.Helper()
	keys, err := w.cx.rel.List(context.Background(), objstore.QueuePrefix("bab", "titogo"))
	if err != nil {
		t.Fatal(err)
	}
	return keys
}

func (w *scanClaimWorld) results(t *testing.T) []string {
	t.Helper()
	keys, err := w.cx.rel.List(context.Background(), objstore.ScanResultsPrefix("bab", "titogo"))
	if err != nil {
		t.Fatal(err)
	}
	return keys
}

// effectsOf returns the app-mutation effects recorded for one action.
func effectsOf(a hostops.AppAdmission, action string) []hostops.Effect {
	var out []hostops.Effect
	for _, e := range a.Effects {
		if e.Resource == hostops.OpAppMutation && e.Action == action {
			out = append(out, e)
		}
	}
	return out
}

func effectStates(es []hostops.Effect) map[string]int {
	out := map[string]int{}
	for _, e := range es {
		out[e.State]++
	}
	return out
}

// requireOpen fails unless the app is open with no operation and no effect
// left issued or uncertain.
func requireOpen(t *testing.T, a hostops.AppAdmission, when string) {
	t.Helper()
	if a.State != "open" || a.Op != nil {
		t.Fatalf("%s: admission state=%q op=%+v, want open with no operation", when, a.State, a.Op)
	}
	for id, e := range a.Effects {
		if e.State != "complete" {
			t.Fatalf("%s: effect %s (%s) is %q, want complete", when, id, e.Action, e.State)
		}
	}
}

// TestClaimScanDrainUncertainScanHoldsOnlyItsGroup: object A carries a verdict
// already (no engine call), object B's Scan fails after the guard marked it
// issued. A's items are gone and A is in done; B's item survives; the app is
// held under B's objstore.DrainGroup operation with exactly one uncertain
// effect, and A's own effect is complete under a different operation. The hold
// is the boundary's honest ruling about a backend call with no known outcome
// (pinned, not hidden), and a second drain refuses before it lists, scans or
// deletes anything.
func TestClaimScanDrainUncertainScanHoldsOnlyItsGroup(t *testing.T) {
	w := newScanClaimWorld(t)
	versionA := w.seed(t, "a.txt")
	w.seed(t, "b.txt")
	w.cx.scanFake.Put(w.bucket, "a.txt", versionA, objstore.VerdictClean)
	base := time.Now().Add(-time.Minute)
	w.enqueue(t, "a.txt", base)
	w.enqueue(t, "b.txt", base.Add(time.Second))
	errBoom := errors.New("clamd went away mid-scan")
	w.cx.scanFake.Fail("Scan", errBoom)

	done, err := w.drain(t)
	if !errors.Is(err, errBoom) {
		t.Fatalf("drain err = %v, want the engine failure", err)
	}
	if len(done) != 1 || done[0].Key != "a.txt" || done[0].Verdict != objstore.VerdictClean {
		t.Fatalf("done = %+v, want A's verdict alone", done)
	}
	left := w.queued(t)
	if len(left) != 1 {
		t.Fatalf("queue holds %v, want only B's item", left)
	}
	if n := w.cx.scanFake.Scans("b.txt"); n != 0 {
		t.Fatalf("B was scanned %d time(s) after its Scan failed", n)
	}

	a := w.admission(t)
	if a.State != "closed" || a.Op == nil || a.Op.Kind != hostops.OpAppMutation || a.Op.Intent["action"] != "objstore.DrainGroup" {
		t.Fatalf("admission state=%q op=%+v, want closed under objstore.DrainGroup", a.State, a.Op)
	}
	groups := effectsOf(a, "objstore.DrainGroup")
	if got := effectStates(groups); got["uncertain"] != 1 || got["complete"] != 1 || len(groups) != 2 {
		t.Fatalf("DrainGroup effects by state = %v (%d), want one complete (A) and one uncertain (B)", got, len(groups))
	}
	for _, e := range groups {
		switch e.State {
		case "uncertain":
			if e.OperationID != a.Op.ID {
				t.Errorf("the uncertain effect belongs to operation %s, but the app is held under %s", e.OperationID, a.Op.ID)
			}
		case "complete":
			if e.OperationID == a.Op.ID {
				t.Error("A's completed effect is recorded under B's held operation: the groups shared one operation")
			}
		}
	}
	if n := len(effectsOf(a, "objstore.DrainApp")); n != 0 {
		t.Fatalf("%d whole-page objstore.DrainApp effect(s) recorded", n)
	}

	// Held means held: the next drain refuses before any read of the object.
	done, err = w.drain(t)
	if !errors.Is(err, hostops.ErrPending) || len(done) != 0 {
		t.Fatalf("second drain = %+v, %v; want ErrPending and no outcome", done, err)
	}
	if n := w.cx.scanFake.Scans("b.txt"); n != 0 {
		t.Fatalf("B was scanned %d time(s) while the app was held", n)
	}
	if got := w.queued(t); len(got) != 1 {
		t.Fatalf("queue = %v after the refused drain, want B's item untouched", got)
	}
}

// TestClaimScanDrainFailureBeforeEffectReopens: B's Status (a read) fails,
// so nothing was issued under B's operation. It settles not-applied, the app
// reopens, B's item is kept, and the next drain scans it.
func TestClaimScanDrainFailureBeforeEffectReopens(t *testing.T) {
	w := newScanClaimWorld(t)
	w.seed(t, "a.txt")
	w.seed(t, "b.txt")
	base := time.Now().Add(-time.Minute)
	w.enqueue(t, "a.txt", base)
	w.enqueue(t, "b.txt", base.Add(time.Second))
	errRead := errors.New("HeadObject: 500")
	// Arm the failure for the SECOND Status call only: A's must succeed.
	var calls int
	var mu sync.Mutex
	w.cx.scanFake.OnStatus(func() {
		mu.Lock()
		defer mu.Unlock()
		calls++
		if calls == 2 {
			w.cx.scanFake.Fail("Status", errRead)
		}
	})

	done, err := w.drain(t)
	if !errors.Is(err, errRead) {
		t.Fatalf("drain err = %v, want the read failure", err)
	}
	if len(done) != 1 || done[0].Key != "a.txt" || !done[0].Scanned {
		t.Fatalf("done = %+v, want A scanned", done)
	}
	if got := w.queued(t); len(got) != 1 {
		t.Fatalf("queue = %v, want B's item kept", got)
	}
	a := w.admission(t)
	requireOpen(t, a, "after a read failure")
	groups := effectsOf(a, "objstore.DrainGroup")
	var notApplied int
	for _, e := range groups {
		if e.ResultReference == "not-applied" {
			notApplied++
		}
	}
	if len(groups) != 2 || notApplied != 1 {
		t.Fatalf("DrainGroup effects = %+v, want A joined and B not-applied", groups)
	}

	w.cx.scanFake.OnStatus(nil)
	done, err = w.drain(t)
	if err != nil || len(done) != 1 || done[0].Key != "b.txt" || !done[0].Scanned {
		t.Fatalf("retry = %+v, %v; want B scanned", done, err)
	}
	if got := w.queued(t); len(got) != 0 {
		t.Fatalf("queue = %v after the retry, want empty", got)
	}
	requireOpen(t, w.admission(t), "after the retry")
}

// TestClaimScanDrainKnownOutcomesAreNotFailures: an object that is gone, a
// consent withdrawn after the engine answered, and a deployment disabled after
// the engine answered are results of a completed effect. None holds the app;
// the queue is dropped (with the results) for the permanent refusal and kept
// for the reversible one.
func TestClaimScanDrainKnownOutcomesAreNotFailures(t *testing.T) {
	t.Run("object gone", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.enqueue(t, "missing.txt", time.Now().Add(-time.Minute))
		done, err := w.drain(t)
		if err != nil {
			t.Fatalf("drain: %v", err)
		}
		if len(done) != 1 || !done[0].Gone || done[0].Key != "missing.txt" {
			t.Fatalf("done = %+v, want one Gone outcome", done)
		}
		if got := w.queued(t); len(got) != 0 {
			t.Fatalf("queue = %v, want the gone object's item deleted", got)
		}
		a := w.admission(t)
		requireOpen(t, a, "after a gone object")
		if n := len(effectsOf(a, "objstore.DrainGroup")); n != 1 {
			t.Fatalf("%d DrainGroup effects, want one", n)
		}
	})

	// Consent withdrawn between the per-object check and the result write:
	// the engine ran (unrecordable, so the drain goes to write a result),
	// the re-read says the store may no longer be scanned. The group's
	// operation completes, and the queue and results are dropped under
	// objstore.DropQueue.
	t.Run("consent withdrawn after the scan", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.seed(t, "a.txt")
		w.seed(t, "b.txt")
		w.cx.scanFake.Unrecordable("a.txt")
		base := time.Now().Add(-time.Minute)
		w.enqueue(t, "a.txt", base)
		w.enqueue(t, "b.txt", base.Add(time.Second))
		if err := objstore.WriteScanResult(context.Background(), w.cx.rel, "bab", "titogo",
			&objstore.ScanResult{Key: "old.txt", Verdict: "clean", Version: "v1", At: time.Now()}); err != nil {
			t.Fatal(err)
		}
		var once sync.Once
		w.cx.scanFake.OnStatus(func() {
			once.Do(func() {
				ctx := context.Background()
				rec, etag, err := objstore.Get(ctx, w.cx.rel, "bab", "titogo")
				if err != nil || rec == nil {
					t.Errorf("re-read record: %+v %v", rec, err)
					return
				}
				rec.Scanning = nil
				if _, err := objstore.Put(ctx, w.cx.rel, "bab", "titogo", rec, etag); err != nil {
					t.Errorf("revoke: %v", err)
				}
			})
		})

		done, err := w.drain(t)
		if !errors.Is(err, objstore.ErrQueueNotPermitted) {
			t.Fatalf("drain err = %v, want ErrQueueNotPermitted", err)
		}
		if len(done) != 0 {
			t.Fatalf("done = %+v for a store that may not be scanned", done)
		}
		if n := w.cx.scanFake.Scans("a.txt"); n != 1 {
			t.Fatalf("A scanned %d time(s), want once (the outcome is about an answer the engine gave)", n)
		}
		if n := w.cx.scanFake.Scans("b.txt"); n != 0 {
			t.Fatalf("B scanned %d time(s) after consent was withdrawn", n)
		}
		if got := w.queued(t); len(got) != 0 {
			t.Fatalf("queue = %v, want dropped", got)
		}
		if got := w.results(t); len(got) != 0 {
			t.Fatalf("results = %v, want dropped with the queue", got)
		}
		a := w.admission(t)
		requireOpen(t, a, "after consent was withdrawn")
		if n := len(effectsOf(a, "objstore.DropQueue")); n != 1 {
			t.Fatalf("%d DropQueue effects, want one", n)
		}
		if got := effectStates(effectsOf(a, "objstore.DrainGroup")); got["complete"] != 1 || len(got) != 1 {
			t.Fatalf("DrainGroup effects by state = %v, want one complete", got)
		}
	})

	t.Run("deployment disabled after the scan", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.seed(t, "a.txt")
		w.seed(t, "b.txt")
		w.cx.scanFake.Unrecordable("a.txt")
		base := time.Now().Add(-time.Minute)
		w.enqueue(t, "a.txt", base)
		w.enqueue(t, "b.txt", base.Add(time.Second))
		var once sync.Once
		w.cx.scanFake.OnStatus(func() {
			once.Do(func() { disableDeploymentScanning(t, w.cx.consoleFixture) })
		})

		done, err := w.drain(t)
		if !errors.Is(err, objstore.ErrScanningDisabled) {
			t.Fatalf("drain err = %v, want ErrScanningDisabled", err)
		}
		if len(done) != 0 {
			t.Fatalf("done = %+v while the deployment is disabled", done)
		}
		if got := w.queued(t); len(got) != 2 {
			t.Fatalf("queue = %v, want both items kept for the next enabled tick", got)
		}
		a := w.admission(t)
		requireOpen(t, a, "after the deployment was disabled")
		if got := effectStates(effectsOf(a, "objstore.DrainGroup")); got["complete"] != 1 || len(got) != 1 {
			t.Fatalf("DrainGroup effects by state = %v, want one complete", got)
		}
		if n := len(effectsOf(a, "objstore.DropQueue")); n != 0 {
			t.Fatalf("%d DropQueue effects for a reversible refusal", n)
		}
	})
}

// TestClaimScanDrainOneOperationPerGroup: three objects, three complete
// objstore.DrainGroup effects under three distinct operations, none uncertain,
// and the app open afterwards.
func TestClaimScanDrainOneOperationPerGroup(t *testing.T) {
	w := newScanClaimWorld(t)
	base := time.Now().Add(-time.Minute)
	for i, key := range []string{"a.txt", "b.txt", "c.txt"} {
		w.seed(t, key)
		w.enqueue(t, key, base.Add(time.Duration(i)*time.Second))
	}
	done, err := w.drain(t)
	if err != nil || len(done) != 3 {
		t.Fatalf("drain = %+v, %v; want three outcomes", done, err)
	}
	a := w.admission(t)
	requireOpen(t, a, "after a full page")
	groups := effectsOf(a, "objstore.DrainGroup")
	ops := map[string]bool{}
	for _, e := range groups {
		if e.State != "complete" {
			t.Errorf("effect %s is %q", e.ID, e.State)
		}
		ops[e.OperationID] = true
	}
	if len(groups) != 3 || len(ops) != 3 {
		t.Fatalf("%d DrainGroup effects under %d operations, want 3 and 3", len(groups), len(ops))
	}
	if got := w.queued(t); len(got) != 0 {
		t.Fatalf("queue = %v, want empty", got)
	}
}

// TestClaimScanHousekeepingOpensOperationsOnlyWithWork: the collects open no
// operation when nothing is expired, and exactly one when something is.
func TestClaimScanHousekeepingOpensOperationsOnlyWithWork(t *testing.T) {
	ctx := context.Background()
	t.Run("results collect", func(t *testing.T) {
		w := newScanClaimWorld(t)
		if err := objstore.WriteScanResult(ctx, w.cx.rel, "bab", "titogo",
			&objstore.ScanResult{Key: "a.txt", Verdict: "clean", Version: "v1", At: time.Now()}); err != nil {
			t.Fatal(err)
		}
		before := w.admission(t)
		n, err := objstore.CollectScanResults(ctx, w.cx.srv.cfg.Releases, "bab", "titogo", time.Now().Add(-objstore.QueueTTL))
		if err != nil || n != 0 {
			t.Fatalf("collect of a fresh result = %d, %v", n, err)
		}
		after := w.admission(t)
		if after.Revision != before.Revision || len(after.Effects) != len(before.Effects) {
			t.Fatalf("admission moved from rev %d (%d effects) to rev %d (%d effects) for a collect with nothing to delete",
				before.Revision, len(before.Effects), after.Revision, len(after.Effects))
		}
		n, err = objstore.CollectScanResults(ctx, w.cx.srv.cfg.Releases, "bab", "titogo", time.Now().Add(time.Hour))
		if err != nil || n != 1 {
			t.Fatalf("collect of an expired result = %d, %v", n, err)
		}
		if got := w.results(t); len(got) != 0 {
			t.Fatalf("results = %v, want collected", got)
		}
		a := w.admission(t)
		requireOpen(t, a, "after the collect")
		if n := len(effectsOf(a, "objstore.CollectScanResults")); n != 1 {
			t.Fatalf("%d CollectScanResults effects, want one", n)
		}
	})

	t.Run("drain collect", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.seed(t, "a.txt")
		w.enqueue(t, "a.txt", time.Now().Add(-time.Minute))
		done, err := w.drain(t)
		if err != nil || len(done) != 1 {
			t.Fatalf("drain = %+v, %v", done, err)
		}
		a := w.admission(t)
		if n := len(effectsOf(a, "objstore.DrainCollect")); n != 0 {
			t.Fatalf("%d DrainCollect effects for a page with nothing stale", n)
		}

		w.seed(t, "b.txt")
		w.enqueue(t, "expired.txt", time.Now().Add(-objstore.QueueTTL-time.Hour))
		w.enqueue(t, "b.txt", time.Now().Add(-time.Minute))
		done, err = w.drain(t)
		if err != nil || len(done) != 1 || done[0].Key != "b.txt" {
			t.Fatalf("drain = %+v, %v; want B alone", done, err)
		}
		if got := w.queued(t); len(got) != 0 {
			t.Fatalf("queue = %v, want the expired item collected and B drained", got)
		}
		a = w.admission(t)
		requireOpen(t, a, "after the collect")
		if got := effectStates(effectsOf(a, "objstore.DrainCollect")); got["complete"] != 1 || len(got) != 1 {
			t.Fatalf("DrainCollect effects by state = %v, want exactly one complete", got)
		}
	})

	t.Run("rebuild collect", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.enqueue(t, "a.txt", time.Now().Add(-time.Minute))
		before := w.admission(t)
		fresh := objstore.NewQueueIndex()
		live, err := objstore.RebuildQueue(ctx, w.cx.srv.cfg.Releases, fresh, "bab", "titogo", time.Now())
		if err != nil || live != 1 || fresh.Len() != 1 {
			t.Fatalf("rebuild = live %d, indexed %d, %v", live, fresh.Len(), err)
		}
		after := w.admission(t)
		if after.Revision != before.Revision || len(after.Effects) != len(before.Effects) {
			t.Fatalf("admission moved for a rebuild with nothing to collect: rev %d -> %d", before.Revision, after.Revision)
		}

		w.enqueue(t, "expired.txt", time.Now().Add(-objstore.QueueTTL-time.Hour))
		fresh = objstore.NewQueueIndex()
		live, err = objstore.RebuildQueue(ctx, w.cx.srv.cfg.Releases, fresh, "bab", "titogo", time.Now())
		if err != nil || live != 1 || fresh.Len() != 1 {
			t.Fatalf("rebuild with an expired item = live %d, indexed %d, %v", live, fresh.Len(), err)
		}
		if got := w.queued(t); len(got) != 1 {
			t.Fatalf("queue = %v, want the live item alone", got)
		}
		a := w.admission(t)
		requireOpen(t, a, "after the rebuild collect")
		if got := effectStates(effectsOf(a, "objstore.RebuildCollect")); got["complete"] != 1 || len(got) != 1 {
			t.Fatalf("RebuildCollect effects by state = %v, want exactly one complete", got)
		}
	})
}

// TestClaimScanDrainRefusesUnderAnotherOwner: with the app closed under a
// foreign app mutation or a purge, the drain refuses before any object is
// read and deletes nothing. And while a DrainGroup is live (the engine is
// mid-Status), a competing exclusive begin on the app refuses: the writeSidecar
// shaped race the queue comments describe is closed by construction.
func TestClaimScanDrainRefusesUnderAnotherOwner(t *testing.T) {
	ctx := context.Background()
	refused := func(t *testing.T, w *scanClaimWorld, what string) {
		t.Helper()
		done, err := w.drain(t)
		if !errors.Is(err, hostops.ErrPending) || len(done) != 0 {
			t.Fatalf("%s: drain = %+v, %v; want ErrPending and nothing done", what, done, err)
		}
		if n := w.cx.scanFake.Scans("a.txt"); n != 0 {
			t.Fatalf("%s: the object was scanned %d time(s)", what, n)
		}
		if got := w.queued(t); len(got) != 1 {
			t.Fatalf("%s: queue = %v, want the item untouched", what, got)
		}
	}

	t.Run("foreign exclusive", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.seed(t, "a.txt")
		w.enqueue(t, "a.txt", time.Now().Add(-time.Minute))
		restore := w.cx.foreign(t, w.ref)
		refused(t, w, "foreign")
		restore()
		done, err := w.drain(t)
		if err != nil || len(done) != 1 {
			t.Fatalf("drain once reopened = %+v, %v", done, err)
		}
	})

	t.Run("purge", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.seed(t, "a.txt")
		w.enqueue(t, "a.txt", time.Now().Add(-time.Minute))
		a := w.admission(t)
		a.Revision++
		a.State = "closed"
		a.Op = &hostops.AppOperation{ID: "purge", Kind: hostops.OpPurging, AppID: a.AppID, Phase: "draining",
			Intent: map[string]string{"account": "bab", "app": "titogo"}}
		w.cx.putAdmission(t, "bab", "titogo", a)
		refused(t, w, "purge")
	})

	t.Run("competing begin while a group is live", func(t *testing.T) {
		w := newScanClaimWorld(t)
		w.seed(t, "a.txt")
		w.enqueue(t, "a.txt", time.Now().Add(-time.Minute))
		var once sync.Once
		var competing error
		var held *hostops.AppAdmission
		w.cx.scanFake.OnStatus(func() {
			once.Do(func() {
				obs, err := hostops.ReadCanonicalAdmission(ctx, w.cx.rel, "bab", "titogo")
				if err != nil {
					t.Errorf("read admission mid-scan: %v", err)
					return
				}
				held = &obs.Value
				_, competing = w.cx.srv.cfg.ClaimAuthority.Control.BeginExclusive(ctx, "bab", "titogo", w.appID,
					hostops.AppOperation{ID: "competing", Kind: hostops.OpAppMutation, AppID: w.appID, Phase: "issued",
						Intent: map[string]string{"action": "competing"}}, claimTestOwner)
			})
		})
		done, err := w.drain(t)
		if err != nil || len(done) != 1 {
			t.Fatalf("drain = %+v, %v", done, err)
		}
		if held == nil || held.State != "closed" || held.Op == nil || held.Op.Intent["action"] != "objstore.DrainGroup" {
			t.Fatalf("mid-Status admission = %+v, want closed under objstore.DrainGroup", held)
		}
		if !errors.Is(competing, hostops.ErrPending) {
			t.Fatalf("competing BeginExclusive during the scan = %v, want ErrPending", competing)
		}
		requireOpen(t, w.admission(t), "after the drain")
	})
}