Fix time-sync and support for non-AppleTV

Author: omarroth

cmd/doubletake/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"log"
@@ -234,7 +235,10 @@ func main() {
 	// for Apple TV compatibility in the normal modern flow.
 	if client.FpEkey == nil {
 		if err := client.FairPlaySetup(ctx); err != nil {
-			log.Fatalf("FairPlay setup failed: %v", err)
+			if !errors.Is(err, airplay.ErrFairPlayUnsupported) {
+				log.Fatalf("FairPlay setup failed: %v", err)
+			}
+			log.Printf("FairPlay SAP unsupported (%v); continuing with pair-verify DataStream setup", err)
 		} else {
 			log.Println("FairPlay setup complete")
 		}

internal/airplay/audio.go

@@ -176,6 +176,47 @@ func (ac *AudioCapture) ReadFrame(buf []byte) (int, error) {
 	return n, nil
 }
 
+// DrainStale discards any PCM that buffered in the OS pipe between capture
+// start and the first read. The capture pipeline starts producing audio
+// immediately, but streaming does not begin until the first video frame is
+// sent; during that gap the kernel pipe accumulates a FIFO backlog that would
+// otherwise be read in order forever, leaving every frame permanently stale and
+// audio lagging video. Draining once just before the read loop starts streaming
+// from the freshest sample. It removes whatever backlog actually accumulated —
+// no fixed latency value is assumed.
+func (ac *AudioCapture) DrainStale() {
+	type deadlineReader interface {
+		SetReadDeadline(t time.Time) error
+	}
+	dr, ok := ac.pcmPipe.(deadlineReader)
+	if !ok {
+		return
+	}
+	buf := make([]byte, 32*1024)
+	var discarded int
+	for {
+		// Re-arm a short idle timeout each read: while a backlog exists, reads
+		// return buffered data immediately; once the pipe is empty the read
+		// blocks and this deadline fires before the next live frame (~8ms)
+		// arrives, ending the drain. This is a poll timeout, not a latency.
+		if err := dr.SetReadDeadline(time.Now().Add(2 * time.Millisecond)); err != nil {
+			break
+		}
+		n, err := ac.pcmPipe.Read(buf)
+		discarded += n
+		if err != nil {
+			break
+		}
+	}
+	// Restore blocking reads for steady-state streaming.
+	_ = dr.SetReadDeadline(time.Time{})
+	if discarded > 0 {
+		const bytesPerSecond = 44100 * 2 * 2 // 44.1kHz, stereo, S16LE
+		dbg("[AUDIO] drained %d bytes (~%.0fms) of startup backlog before streaming",
+			discarded, float64(discarded)/bytesPerSecond*1000)
+	}
+}
+
 func (ac *AudioCapture) Stop() {
 	if ac.stopped {
 		return
@@ -413,7 +454,7 @@ func (s *MirrorSession) setupAudioStream(dataPort, controlPort int, aesKey, aesI
 			as.chachaNonceMode.String(), as.chachaAADMode.String())
 	}
 	if latencyOverride > 0 {
-		dbg("[AUDIO] receiver audio latency override: %d samples", latencySamples)
+		dbg("[AUDIO] audio latency: %d samples", latencySamples)
 	}
 	dbg("[AUDIO] local ports: data=%d (→remote %d) ctrl=%d (→remote %d)",
 		dataLocalPort, dataPort, ctrlLocalPort, controlPort)
@@ -594,12 +635,18 @@ func (as *AudioStream) sendSyncPacket(ntpTime uint64, isFirst bool) error {
 	latencySamples := as.latencySamples
 	as.mu.Unlock()
 
+	// anchorLatency is the playout lead time reported to the receiver: the newest
+	// audio we have sent (rtpNow) plays anchorLatency/44100 seconds after "now".
+	// This equals the negotiated session latency, the same forward bias video
+	// frames carry, so audio and video captured at the same instant play together.
+	anchorLatency := latencySamples
+
 	// Sync packet: 20 bytes total (8-byte RTP-like header + 12-byte payload)
 	// Format observed from real Apple senders:
 	//   header: V=2, X=1(first)/0(subsequent), M=1, PT=84, seq=4 (constant)
 	//   RTP timestamp = current playback position (sync_rtp)
 	//   payload: NTP_hi(4) + NTP_lo(4) + next_rtp(4)
-	//   next_rtp = sync_rtp + latencySamples
+	//   next_rtp = current receive head (rtpNow)
 	packet := make([]byte, 20)
 	if isFirst {
 		packet[0] = 0x90 // V=2, X=1
@@ -609,16 +656,16 @@ func (as *AudioStream) sendSyncPacket(ntpTime uint64, isFirst bool) error {
 	packet[1] = 0xd4 // M=1, PT=84
 	// seq field is constant 4 in working pcap captures
 	binary.BigEndian.PutUint16(packet[2:4], 4)
-	// Bytes 4-7: sync_rtp = current playback position
+	// Bytes 4-7: sync_rtp = current playback position = receive head - anchorLatency
 	syncRtp := rtpNow
-	if rtpNow >= latencySamples {
-		syncRtp = rtpNow - latencySamples
+	if rtpNow >= anchorLatency {
+		syncRtp = rtpNow - anchorLatency
 	}
 	binary.BigEndian.PutUint32(packet[4:8], syncRtp)
 	// Bytes 8-15: NTP timestamp (current wall-clock time)
 	binary.BigEndian.PutUint64(packet[8:16], ntpTime)
-	// Bytes 16-19: next_rtp = sync_rtp + latencySamples
-	binary.BigEndian.PutUint32(packet[16:20], syncRtp+latencySamples)
+	// Bytes 16-19: next_rtp = current receive head
+	binary.BigEndian.PutUint32(packet[16:20], rtpNow)
 
 	_, err := as.ctrlConn.WriteTo(packet, as.ctrlAddr)
 	return err
@@ -744,6 +791,11 @@ func (s *MirrorSession) StreamAudio(ctx context.Context, capture *AudioCapture,
 	burstDone := false
 	frameBuf := make([]byte, 8192)
 
+	// Capture started before video did, so the OS pipe holds a backlog of stale
+	// audio accumulated while we waited for the first video frame. Drop it so we
+	// begin streaming from the freshest sample and audio lines up with video.
+	capture.DrainStale()
+
 	for {
 		select {
 		case <-ctx.Done():

internal/airplay/client.go

@@ -24,6 +24,7 @@ import (
 type ReceiverInfo struct {
 	Name              string  `plist:"name"`
 	Model             string  `plist:"model"`
+	Manufacturer      string  `plist:"manufacturer"`
 	DeviceID          string  `plist:"deviceID"`
 	ProtocolVersion   string  `plist:"protocolVersion"`
 	SourceVersion     string  `plist:"sourceVersion"`
@@ -40,6 +41,16 @@ type ReceiverInfo struct {
 	MacAddress        string  `plist:"macAddress"`
 }
 
+// HTTPStatusError is returned when a receiver responds with a non-2xx RTSP/HTTP status.
+type HTTPStatusError struct {
+	StatusCode int
+	Body       []byte
+}
+
+func (e *HTTPStatusError) Error() string {
+	return fmt.Sprintf("HTTP %d (body: %s)", e.StatusCode, string(e.Body))
+}
+
 // AirPlayClient manages the connection to an AirPlay receiver.
 type AirPlayClient struct {
 	host string
@@ -322,7 +333,7 @@ func (c *AirPlayClient) readPlaintextHTTPResponse() ([]byte, map[string]string,
 			io.ReadFull(c.conn, errBody)
 		}
 		dbg("[READ] error response body (%d bytes): %s", len(errBody), hex.EncodeToString(errBody))
-		return nil, headers, fmt.Errorf("HTTP %d (body: %s)", statusCode, string(errBody))
+		return nil, headers, &HTTPStatusError{StatusCode: statusCode, Body: errBody}
 	}
 
 	if contentLength == 0 {
@@ -390,7 +401,7 @@ func (c *AirPlayClient) readEncryptedHTTPResponse() ([]byte, map[string]string,
 			remaining = remaining[:contentLength]
 		}
 		dbg("[ENC-READ] error response body (%d bytes): %s", len(remaining), hex.EncodeToString(remaining))
-		return nil, headers, fmt.Errorf("HTTP %d (body: %s)", statusCode, string(remaining))
+		return nil, headers, &HTTPStatusError{StatusCode: statusCode, Body: remaining}
 	}
 
 	if contentLength == 0 {

internal/airplay/discovery.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/grandcat/zeroconf"
 )
@@ -152,3 +153,23 @@ func (d *AirPlayDevice) SupportsScreen() bool {
 func (d *AirPlayDevice) SupportsTransientPairing() bool {
 	return d.Features&FeatureTransientPairing != 0
 }
+
+func (d *AirPlayDevice) SupportsFairPlaySAP() bool {
+	return d.Features&FeatureFPSAP25 != 0
+}
+
+func (i *ReceiverInfo) SupportsFairPlaySAP() bool {
+	return i != nil && i.Features&FeatureFPSAP25 != 0
+}
+
+// playoutLatencyFloor returns the minimum playout lead this receiver needs.
+// Modern Apple receivers advertise FairPlay SAP and have robust audio jitter
+// buffers, so they can play at very low latency (floor 0). Receivers without it
+// (Roku and other third-party AirPlay implementations) need a conservative lead
+// or they drop audio they can no longer schedule.
+func (i *ReceiverInfo) playoutLatencyFloor() time.Duration {
+	if i != nil && i.SupportsFairPlaySAP() {
+		return 0
+	}
+	return conservativePlayoutLatency
+}

internal/airplay/discovery_test.go

@@ -33,3 +33,21 @@ func TestUnescapeDNSName(t *testing.T) {
 		})
 	}
 }
+
+func TestSupportsFairPlaySAP(t *testing.T) {
+	rokuFeatures := uint64(0x38bcf46007f8ad0)
+	if (&ReceiverInfo{Features: rokuFeatures}).SupportsFairPlaySAP() {
+		t.Fatalf("Roku feature mask unexpectedly advertises FPSAP")
+	}
+	if (&AirPlayDevice{Features: rokuFeatures}).SupportsFairPlaySAP() {
+		t.Fatalf("Roku discovery feature mask unexpectedly advertises FPSAP")
+	}
+
+	withFairPlay := rokuFeatures | FeatureFPSAP25
+	if !(&ReceiverInfo{Features: withFairPlay}).SupportsFairPlaySAP() {
+		t.Fatalf("ReceiverInfo with FPSAP bit did not advertise FairPlay SAP")
+	}
+	if !(&AirPlayDevice{Features: withFairPlay}).SupportsFairPlaySAP() {
+		t.Fatalf("AirPlayDevice with FPSAP bit did not advertise FairPlay SAP")
+	}
+}

internal/airplay/fairplay.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"crypto/sha512"
 	"encoding/hex"
+	"errors"
 	"fmt"
 
 	"doubletake/internal/fpemu"
@@ -13,6 +14,8 @@ import (
 // fairPlayM1 is the fixed m1 blob that matches the snapshot state.
 var fairPlayM1 = mustDecodeHexFP("46504c590301010000000004020003bb")
 
+var ErrFairPlayUnsupported = errors.New("receiver does not support FairPlay SAP")
+
 func mustDecodeHexFP(s string) []byte {
 	b, err := hex.DecodeString(s)
 	if err != nil {
@@ -24,6 +27,10 @@ func mustDecodeHexFP(s string) []byte {
 // FairPlaySetup performs the complete FairPlay SAP handshake using the
 // standalone ARM64 interpreter.
 func (c *AirPlayClient) FairPlaySetup(ctx context.Context) error {
+	if c.info != nil && !c.info.SupportsFairPlaySAP() {
+		return fmt.Errorf("%w: FPSAP feature bit is not advertised (features=0x%x)", ErrFairPlayUnsupported, c.info.Features)
+	}
+
 	dbg("[FP] starting FairPlay SAP handshake...")
 
 	// Phase 1: Send m1, receive m2
@@ -34,6 +41,10 @@ func (c *AirPlayClient) FairPlaySetup(ctx context.Context) error {
 	m2, err := c.httpRequest("POST", "/fp-setup", "application/octet-stream", m1,
 		map[string]string{"X-Apple-ET": "32"})
 	if err != nil {
+		var statusErr *HTTPStatusError
+		if errors.As(err, &statusErr) && statusErr.StatusCode == 404 {
+			return fmt.Errorf("%w: /fp-setup returned 404", ErrFairPlayUnsupported)
+		}
 		return fmt.Errorf("fp-setup phase 1 (m1): %w", err)
 	}
 
@@ -61,6 +72,10 @@ func (c *AirPlayClient) FairPlaySetup(ctx context.Context) error {
 	m4, err := c.httpRequest("POST", "/fp-setup", "application/octet-stream", m3,
 		map[string]string{"X-Apple-ET": "32"})
 	if err != nil {
+		var statusErr *HTTPStatusError
+		if errors.As(err, &statusErr) && statusErr.StatusCode == 404 {
+			return fmt.Errorf("%w: /fp-setup returned 404 during phase 2", ErrFairPlayUnsupported)
+		}
 		return fmt.Errorf("fp-setup phase 2 (m3): %w", err)
 	}
 

internal/airplay/latency.go

@@ -6,7 +6,19 @@ import (
 	"time"
 )
 
-const defaultTargetLatency = 100 * time.Millisecond
+const defaultTargetLatency = 1 * time.Millisecond
+
+// conservativePlayoutLatency is the playout lead required by receivers that lack
+// a robust audio jitter buffer (third-party AirPlay implementations such as
+// Roku, which do not advertise FairPlay SAP). The control-port sync anchor
+// reports that the newest audio frame plays this far in the future, which is
+// also the buffer lead the receiver has to schedule each packet before its play
+// time. With too little lead these receivers drop audio they can no longer
+// schedule. Modern Apple receivers buffer aggressively and do not need this, so
+// it is applied per-receiver (see ReceiverInfo.playoutLatencyFloor), not
+// globally — audio and video share whatever latency is chosen so they stay in
+// sync.
+const conservativePlayoutLatency = 500 * time.Millisecond
 
 var targetLatencyNS atomic.Int64
 
@@ -36,7 +48,10 @@ func TargetLatency() time.Duration {
 }
 
 func targetLatencySamples44k1() uint32 {
-	d := TargetLatency()
+	return samplesFor44k1(TargetLatency())
+}
+
+func samplesFor44k1(d time.Duration) uint32 {
 	samples := int64(math.Round(float64(d) * 44100.0 / float64(time.Second)))
 	if samples < 1 {
 		samples = 1