groundstation/internal/osd/osd.go

// Package osd is the shared flight OSD snapshot contract between the Ground
// Station supervisor and the kiosk display process. The supervisor owns the
// RC, arming, power and video state and publishes immutable snapshots;
// the display renders the latest fresh snapshot on the transparent OSD
// overlay. The snapshot is a plain value: everything published is copied, so
// a reader can never observe or mutate a half-written state.
package osd

import (
	"encoding/binary"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"git.piksar.eu/piotr/groundstation/internal/protocol"
)

const (
	Magic0  byte = 'G'
	Magic1  byte = 'O'
	Version byte = 1

	// Size is the fixed encoded snapshot length: magic and version, frame
	// sequence and timestamp, all 16 final RC channels, one flags byte,
	// the publish time and the CRC.
	Size = 3 + 2 + 4 + protocol.Channels*2 + 1 + 8 + 2

	// DefaultStatePath is where the supervisor publishes snapshots and
	// where the display reads them unless configured otherwise. It lives
	// in tmpfs: snapshots are runtime telemetry, not persistent state.
	DefaultStatePath = "/run/groundstation/osd-state.bin"

	// Heartbeat is how often an unchanged snapshot is republished, so the
	// display keeps seeing it as fresh while the UART stays healthy.
	Heartbeat = 500 * time.Millisecond

	// writeInterval is the minimum gap between two state-file writes. It
	// bounds the write rate when RC channels change on every frame without
	// delaying an arm or power transition noticeably.
	writeInterval = 50 * time.Millisecond

	// Freshness is the display-side maximum age of a rendered snapshot: a
	// dead UART must end OSD rendering instead of freezing stale telemetry
	// over live video.
	Freshness = 2 * time.Second
)

// Snapshot flag bits in the encoded record.
const (
	flagArmed byte = 1 << iota
	flagPower
	flagVideo
)

var (
	ErrLength  = errors.New("groundstation/osd: invalid snapshot length")
	ErrMagic   = errors.New("groundstation/osd: invalid snapshot magic")
	ErrVersion = errors.New("groundstation/osd: unsupported snapshot version")
	ErrCRC     = errors.New("groundstation/osd: invalid snapshot crc")
)

// Snapshot is one immutable flight state: the final RC channels, the arm
// state derived from the configured arm channel, the requested CH6 WFB power
// and the supervisor's decoded-video presence knowledge — its video
// launcher ownership, the only decoded-video signal the supervisor has —
// plus the provenance of the RC frame it was built from. The display never
// renders from this field: it gates on the video lifecycle it observes on
// the display server itself. PublishedMS is stamped at publication and is
// what the display uses to reject stale snapshots.
type Snapshot struct {
	Sequence    uint16
	TimestampMS uint32
	Channels    [protocol.Channels]int16
	Armed       bool
	Power       bool
	Video       bool
	PublishedMS int64
}

func Encode(snapshot Snapshot) []byte {
	buf := make([]byte, Size)
	buf[0], buf[1], buf[2] = Magic0, Magic1, Version
	binary.LittleEndian.PutUint16(buf[3:5], snapshot.Sequence)
	binary.LittleEndian.PutUint32(buf[5:9], snapshot.TimestampMS)
	for i, value := range snapshot.Channels {
		binary.LittleEndian.PutUint16(buf[9+i*2:9+i*2+2], uint16(value))
	}
	var flags byte
	if snapshot.Armed {
		flags |= flagArmed
	}
	if snapshot.Power {
		flags |= flagPower
	}
	if snapshot.Video {
		flags |= flagVideo
	}
	buf[41] = flags
	binary.LittleEndian.PutUint64(buf[42:50], uint64(snapshot.PublishedMS))
	binary.LittleEndian.PutUint16(buf[Size-2:], protocol.CRC16CCITT(buf[:Size-2]))
	return buf
}

func Decode(buf []byte) (Snapshot, error) {
	var snapshot Snapshot
	if len(buf) != Size {
		return snapshot, ErrLength
	}
	if buf[0] != Magic0 || buf[1] != Magic1 {
		return snapshot, ErrMagic
	}
	if buf[2] != Version {
		return snapshot, ErrVersion
	}
	if got, want := binary.LittleEndian.Uint16(buf[Size-2:]), protocol.CRC16CCITT(buf[:Size-2]); got != want {
		return snapshot, fmt.Errorf("%w: got %04x want %04x", ErrCRC, got, want)
	}
	snapshot.Sequence = binary.LittleEndian.Uint16(buf[3:5])
	snapshot.TimestampMS = binary.LittleEndian.Uint32(buf[5:9])
	for i := range snapshot.Channels {
		snapshot.Channels[i] = int16(binary.LittleEndian.Uint16(buf[9+i*2 : 9+i*2+2]))
	}
	flags := buf[41]
	snapshot.Armed = flags&flagArmed != 0
	snapshot.Power = flags&flagPower != 0
	snapshot.Video = flags&flagVideo != 0
	snapshot.PublishedMS = int64(binary.LittleEndian.Uint64(buf[42:50]))
	return snapshot, nil
}

// DecodeFile reads and decodes one snapshot from its state file.
func DecodeFile(path string) (Snapshot, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return Snapshot{}, err
	}
	return Decode(data)
}

// Publisher publishes immutable snapshots latest-wins. Publish never blocks:
// it stores the snapshot atomically and only wakes the writer, so the UART,
// WFB and video paths never wait on state-file I/O. The writer persists the
// latest snapshot whenever it changes and republishes it as a heartbeat so
// the display keeps seeing it as fresh.
type Publisher struct {
	latest        atomic.Pointer[Snapshot]
	wake          chan struct{}
	done          chan struct{}
	stopped       chan struct{}
	closeOnce     sync.Once
	path          string
	heartbeat     time.Duration
	writeInterval time.Duration
	write         func(path string, data []byte) error
	logger        *log.Logger
}

// NewPublisher builds an in-memory publisher without a writer goroutine.
func NewPublisher() *Publisher {
	return newFilePublisher("", nil, Heartbeat, writeInterval, nil)
}

// NewFilePublisher builds a publisher that persists snapshots to path for
// the display process. Failures are logged and retried; they never block
// Publish.
func NewFilePublisher(path string, logger *log.Logger) *Publisher {
	return newFilePublisher(path, writeStateFile, Heartbeat, writeInterval, logger)
}

func newFilePublisher(path string, write func(string, []byte) error, heartbeat, interval time.Duration, logger *log.Logger) *Publisher {
	p := &Publisher{
		wake:          make(chan struct{}, 1),
		done:          make(chan struct{}),
		stopped:       make(chan struct{}),
		path:          path,
		heartbeat:     heartbeat,
		writeInterval: interval,
		write:         write,
		logger:        logger,
	}
	if path != "" {
		go p.loop()
	}
	return p
}

// Publish records the snapshot and wakes the writer. The published value is
// stamped with the publication time and stored as an immutable copy.
func (p *Publisher) Publish(snapshot Snapshot) {
	snapshot.PublishedMS = time.Now().UnixMilli()
	p.latest.Store(&snapshot)
	if p.path == "" {
		return
	}
	select {
	case p.wake <- struct{}{}:
	default:
	}
}

// Latest reports the most recently published snapshot, if any.
func (p *Publisher) Latest() (Snapshot, bool) {
	snapshot := p.latest.Load()
	if snapshot == nil {
		return Snapshot{}, false
	}
	return *snapshot, true
}

// Close stops the writer and removes the state file, so a stopped
// supervisor never leaves stale telemetry for the display to render.
// It is safe to call more than once and on an in-memory publisher.
func (p *Publisher) Close() {
	p.closeOnce.Do(func() {
		if p.path == "" {
			return
		}
		close(p.done)
		<-p.stopped
		_ = os.Remove(p.path)
	})
}

func (p *Publisher) loop() {
	defer close(p.stopped)
	ticker := time.NewTicker(p.writeInterval)
	defer ticker.Stop()

	var last Snapshot
	have := false
	var lastWrite time.Time
	var lastErrLog time.Time
	for {
		select {
		case <-p.wake:
		case <-ticker.C:
		case <-p.done:
			return
		}

		snapshot := p.latest.Load()
		if snapshot == nil {
			continue
		}
		// Changed states go out as soon as the write floor allows; an
		// unchanged state goes out on the heartbeat so the display keeps
		// seeing it as fresh.
		now := time.Now()
		if have {
			if *snapshot == last {
				if now.Sub(lastWrite) < p.heartbeat {
					continue
				}
			} else if now.Sub(lastWrite) < p.writeInterval {
				continue
			}
		}
		if err := p.write(p.path, Encode(*snapshot)); err != nil {
			// Failures are retried on every tick, so a persistently
			// broken state path is logged at most once per heartbeat
			// instead of flooding the journal.
			if p.logger != nil && now.Sub(lastErrLog) >= p.heartbeat {
				p.logger.Printf("osd state write %s: %v", p.path, err)
				lastErrLog = now
			}
			continue
		}
		last, have, lastWrite = *snapshot, true, now
	}
}

// writeStateFile replaces the state file atomically: readers either see the
// previous complete snapshot or the new one, never a partial write. The file
// stays world-readable so the unprivileged display user can read it; it
// carries only flight telemetry.
func writeStateFile(path string, data []byte) error {
	dir := filepath.Dir(path)
	if err := os.MkdirAll(dir, 0755); err != nil {
		return err
	}
	temp, err := os.CreateTemp(dir, filepath.Base(path)+".tmp-*")
	if err != nil {
		return err
	}
	tmpPath := temp.Name()
	defer os.Remove(tmpPath)
	if _, err := temp.Write(data); err != nil {
		temp.Close()
		return err
	}
	if err := temp.Chmod(0644); err != nil {
		temp.Close()
		return err
	}
	if err := temp.Close(); err != nil {
		return err
	}
	return os.Rename(tmpPath, path)
}

// FileSource reads the supervisor-published snapshot from its state file.
// It is the display-side end of the snapshot contract: reads that fail or
// return stale snapshots report no snapshot, so the OSD stays transparent
// instead of rendering telemetry from a dead UART.
type FileSource struct {
	path string
	// now is the clock used for freshness checks; tests replace it.
	now func() time.Time
}

// NewFileSource builds a source reading the given state file.
func NewFileSource(path string) *FileSource {
	return &FileSource{path: path, now: time.Now}
}

// Latest reports the current snapshot while it is fresh.
func (s *FileSource) Latest() (Snapshot, bool) {
	snapshot, err := DecodeFile(s.path)
	if err != nil {
		return Snapshot{}, false
	}
	if age := s.now().Sub(time.UnixMilli(snapshot.PublishedMS)); age > Freshness {
		return Snapshot{}, false
	}
	return snapshot, true
}