//go:build linux

package hostexec

import (
	"context"
	"errors"
	"testing"
	"time"
)

func TestPlan14ExecutionShutdownJoinsAcquisition(t *testing.T) {
	b, disk, remote, _, q, peer := brokerFixture(t)
	entered, resume := make(chan struct{}), make(chan struct{})
	remote.onPut = func(int) { close(entered); <-resume }
	result := make(chan error, 1)
	go func() { _, err := b.Execute(t.Context(), peer, q); result <- err }()
	<-entered
	closing := make(chan error, 1)
	go func() { closing <- b.Close(t.Context()) }()
	var early bool
	select {
	case <-closing:
		early = true
	case <-time.After(50 * time.Millisecond):
	}
	probe, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
	unlock, lockErr := disk.Lock(probe, q.Host+".lock")
	cancel()
	if lockErr == nil {
		unlock()
	}
	close(resume)
	if err := <-result; err == nil {
		t.Error("acquisition returned a permit during shutdown")
	}
	if !early {
		if err := <-closing; err != nil {
			t.Fatal(err)
		}
	}
	if early || !errors.Is(lockErr, context.DeadlineExceeded) {
		t.Fatalf("shutdown released ownership before acquisition joined: early=%v lock=%v", early, lockErr)
	}
	r, err := b.read(q.Host)
	if err != nil || r.Status != "lost" || r.LeaseETag == "" {
		t.Fatalf("late acquisition evidence was lost: %+v %v", r, err)
	}
	probe, cancel = context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	unlock, err = disk.Lock(probe, q.Host+".lock")
	if err != nil {
		t.Fatal(err)
	}
	unlock()
}

func TestPlan14ExecutionShutdownRevokesBeforeCleanup(t *testing.T) {
	b, disk, _, _, q, peer := brokerFixture(t)
	if _, err := b.Execute(t.Context(), peer, q); err != nil {
		t.Fatal(err)
	}
	entered, resume := make(chan struct{}), make(chan struct{})
	b.cfg.Cleanup = func(context.Context, ExecutionRecord) error { close(entered); <-resume; return nil }
	closing := make(chan error, 1)
	go func() { closing <- b.Close(t.Context()) }()
	<-entered
	held, err := disk.LiveHeld(q.Host + ".live")
	close(resume)
	if closeErr := <-closing; closeErr != nil {
		t.Fatal(closeErr)
	}
	if err != nil || held {
		t.Fatalf("cleanup began before live permission was withdrawn: %v", err)
	}
	q.ID = "after-shutdown"
	if _, err := b.Execute(t.Context(), peer, q); err == nil {
		t.Fatal("closed broker accepted new work")
	}
}

func TestPlan14ExecutionShutdownTimeoutRetainsOwnership(t *testing.T) {
	b, disk, remote, _, q, peer := brokerFixture(t)
	entered, resume := make(chan struct{}), make(chan struct{})
	remote.onPut = func(int) { close(entered); <-resume }
	result := make(chan error, 1)
	go func() { _, err := b.Execute(t.Context(), peer, q); result <- err }()
	<-entered
	bounded, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
	err := b.Close(bounded)
	cancel()
	if !errors.Is(err, context.DeadlineExceeded) {
		close(resume)
		<-result
		t.Fatalf("shutdown did not retain the outstanding acquisition: %v", err)
	}
	probe, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
	unlock, lockErr := disk.Lock(probe, q.Host+".lock")
	cancel()
	if lockErr == nil {
		unlock()
	}
	close(resume)
	if err := <-result; err == nil {
		t.Fatal("late acquisition returned permission")
	}
	if err := b.Close(t.Context()); err != nil {
		t.Fatal(err)
	}
	if !errors.Is(lockErr, context.DeadlineExceeded) {
		t.Fatalf("timeout abandoned ownership: %v", lockErr)
	}
}