groundstation/internal/osd/publisher_test.go

package osd

import (
	"bytes"
	"errors"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func TestPublisherKeepsOnlyTheLatestSnapshot(t *testing.T) {
	publisher := NewPublisher()

	if _, ok := publisher.Latest(); ok {
		t.Fatal("a publisher without snapshots must report none")
	}

	first := testSnapshot()
	publisher.Publish(first)
	got, ok := publisher.Latest()
	if !ok {
		t.Fatal("a published snapshot must be readable")
	}
	if got.Sequence != first.Sequence || got.PublishedMS == 0 {
		t.Fatalf("latest=%+v must carry the published state with a publish timestamp", got)
	}

	second := first
	second.Sequence = 7
	second.Video = false
	publisher.Publish(second)
	got, _ = publisher.Latest()
	if got.Sequence != 7 || got.Video {
		t.Fatalf("latest=%+v must be the most recent snapshot", got)
	}
}

func TestPublishNeverBlocksOnASlowWriter(t *testing.T) {
	release := make(chan struct{})
	var writes atomic.Int32
	publisher := newFilePublisher(filepath.Join(t.TempDir(), "osd-state.bin"), func(string, []byte) error {
		writes.Add(1)
		<-release
		return errors.New("unreachable")
	}, Heartbeat, time.Millisecond, nil)

	done := make(chan struct{})
	go func() {
		defer close(done)
		for i := 0; i < 100; i++ {
			snapshot := testSnapshot()
			snapshot.Sequence = uint16(i)
			publisher.Publish(snapshot)
		}
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("publish must never wait for the state file writer")
	}

	close(release)
}

func TestFilePublisherPublishesSnapshotsToTheStateFile(t *testing.T) {
	path := filepath.Join(t.TempDir(), "run", "osd-state.bin")
	publisher := NewFilePublisher(path, nil)
	defer publisher.Close()

	want := testSnapshot()
	publisher.Publish(want)

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if got, err := DecodeFile(path); err == nil && got == want {
			if info, statErr := os.Stat(path); statErr != nil || info.Mode().Perm() != 0644 {
				t.Fatalf("state file mode=%v want 0644 for the unprivileged display user", info.Mode())
			}
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatal("the published snapshot never reached the state file")
}

func TestFilePublisherRewritesUnchangedStateAsHeartbeat(t *testing.T) {
	var writes sync.Mutex
	count := 0
	publisher := newFilePublisher(filepath.Join(t.TempDir(), "osd-state.bin"),
		func(path string, data []byte) error {
			if err := os.WriteFile(path, data, 0644); err != nil {
				return err
			}
			writes.Lock()
			count++
			writes.Unlock()
			return nil
		}, 10*time.Millisecond, 5*time.Millisecond, nil)

	publisher.Publish(testSnapshot())
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		writes.Lock()
		got := count
		writes.Unlock()
		if got >= 3 {
			publisher.Close()
			return
		}
		time.Sleep(time.Millisecond)
	}
	publisher.Close()
	t.Fatal("an unchanged snapshot must keep being republished so it stays fresh")
}

func TestFilePublisherCloseRemovesTheStateFile(t *testing.T) {
	path := filepath.Join(t.TempDir(), "osd-state.bin")
	publisher := NewFilePublisher(path, nil)
	publisher.Publish(testSnapshot())

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(path); err == nil {
			break
		}
		time.Sleep(time.Millisecond)
	}

	publisher.Close()
	publisher.Close() // must be idempotent
	if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("state file stat=%v want removed after close", err)
	}
}

func TestFilePublisherKeepsRetryingAfterAWriteFailure(t *testing.T) {
	var attempts atomic.Int32
	path := filepath.Join(t.TempDir(), "osd-state.bin")
	publisher := newFilePublisher(path, func(path string, data []byte) error {
		if attempts.Add(1) == 1 {
			return errors.New("transient write failure")
		}
		return os.WriteFile(path, data, 0644)
	}, time.Second, 5*time.Millisecond, log.New(os.Stderr, "", 0))
	defer publisher.Close()

	publisher.Publish(testSnapshot())
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if _, err := DecodeFile(path); err == nil {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatal("a failed state write must be retried, not dropped")
}

func TestFilePublisherThrottlesPersistentFailureLogs(t *testing.T) {
	var logs sync.Mutex
	var buffer bytes.Buffer
	publisher := newFilePublisher(filepath.Join(t.TempDir(), "missing-dir", "osd-state.bin"),
		func(string, []byte) error {
			return errors.New("state path is broken")
		}, 20*time.Millisecond, 5*time.Millisecond, log.New(&buffer, "", 0))

	publisher.Publish(testSnapshot())
	time.Sleep(150 * time.Millisecond)
	publisher.Close()

	logs.Lock()
	defer logs.Unlock()
	lines := strings.Count(buffer.String(), "\n")
	// The write retries on every tick but must log at most once per
	// heartbeat: about 150ms/20ms plus the first line, never per attempt.
	if lines == 0 || lines > 8 {
		t.Fatalf("failure log lines=%d want throttled reporting (1..8)", lines)
	}
}