bee/audit/internal/app/blackbox_kick_test.go

package app

import (
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func TestRequestBlackboxSyncUpdatesKickModTime(t *testing.T) {
	dir := t.TempDir()

	before := blackboxKickModTime(dir)
	if !before.IsZero() {
		t.Fatalf("blackboxKickModTime before any request = %v, want zero", before)
	}

	requestBlackboxSync(dir)
	first := blackboxKickModTime(dir)
	if first.IsZero() {
		t.Fatalf("blackboxKickModTime after request is still zero")
	}

	time.Sleep(5 * time.Millisecond)
	requestBlackboxSync(dir)
	second := blackboxKickModTime(dir)
	if !second.After(first) {
		t.Fatalf("second kick mtime %v did not advance past first %v", second, first)
	}
}

func newTestBlackboxWorker(t *testing.T, exportDir string) (*blackboxWorker, *int32) {
	t.Helper()
	rt := &blackboxRuntime{
		exportDir:  exportDir,
		statePath:  filepath.Join(t.TempDir(), "blackbox-state.json"),
		bootFolder: "boot-folder",
		workers:    make(map[string]*blackboxWorker),
	}
	w := &blackboxWorker{
		runtime:     rt,
		flushPeriod: blackboxMinFlushPeriod,
		status:      "running",
		stopCh:      make(chan struct{}),
		stoppedCh:   make(chan struct{}),
	}
	var calls int32
	w.syncCycleFunc = func() error {
		atomic.AddInt32(&calls, 1)
		return nil
	}
	return w, &calls
}

func TestBlackboxWorkerWakesEarlyOnKick(t *testing.T) {
	exportDir := t.TempDir()
	old := blackboxKickPollInterval
	blackboxKickPollInterval = 5 * time.Millisecond
	t.Cleanup(func() { blackboxKickPollInterval = old })

	w, calls := newTestBlackboxWorker(t, exportDir)
	// A long flush period: without the kick mechanism, a second sync
	// wouldn't happen within the test's timeout.
	w.flushPeriod = 10 * time.Second

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		w.run()
	}()

	// Let the first (immediate) cycle happen.
	deadline := time.After(2 * time.Second)
	for atomic.LoadInt32(calls) < 1 {
		select {
		case <-deadline:
			t.Fatal("timed out waiting for first sync cycle")
		case <-time.After(time.Millisecond):
		}
	}

	// Keep requesting a sync until a second cycle happens. A single kick can
	// legitimately land in the brief window where the worker is still
	// recording its post-cycle lastKickSeen baseline and get folded into it
	// harmlessly (the point being tested — that *a* kick wakes the worker
	// well before the 10s flush period — still holds as long as one of these
	// repeated kicks lands after that baseline is recorded).
	deadline = time.After(2 * time.Second)
	for atomic.LoadInt32(calls) < 2 {
		requestBlackboxSync(exportDir)
		select {
		case <-deadline:
			t.Fatalf("timed out waiting for kick-triggered sync cycle; calls=%d", atomic.LoadInt32(calls))
		case <-time.After(5 * time.Millisecond):
		}
	}

	w.stop()
	wg.Wait()
}

func TestBlackboxWorkerStopsCleanly(t *testing.T) {
	exportDir := t.TempDir()
	w, calls := newTestBlackboxWorker(t, exportDir)
	w.flushPeriod = time.Hour

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		w.run()
	}()

	deadline := time.After(2 * time.Second)
	for atomic.LoadInt32(calls) < 1 {
		select {
		case <-deadline:
			t.Fatal("timed out waiting for first sync cycle")
		case <-time.After(time.Millisecond):
		}
	}

	done := make(chan struct{})
	go func() {
		w.stop()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("worker.stop() did not return in time")
	}
	wg.Wait()
}