// Command capture records raw Pro DJ Link UDP datagrams to a JSONL fixture.
//
// It listens on the DJ Link discovery, beat, and status ports and writes one
// Packet per line so tests can replay real traffic without hardware. With
// -announce it first brings a virtual CDJ online (the full startup handshake)
// so real players send status, metadata, and database traffic to it.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

// Packet is one recorded UDP datagram.
type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	announce := flag.Bool("announce", false, "bring a virtual CDJ online while capturing")
	device := flag.Int("device", 5, "virtual CDJ device number")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "capture.jsonl", "output JSONL fixture path")
	secs := flag.Int("secs", 8, "capture duration in seconds")
	flag.Parse()

	if err := run(*out, time.Duration(*secs)*time.Second, *announce, byte(*device), *name); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

func run(out string, d time.Duration, announce bool, device byte, name string) error {
	f, err := os.Create(out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, d)
	defer cancel()

	c0, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50000})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}

	c1, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50001})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c1.Close()

	c2, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50002})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c2.Close()

	var vcdj *link.VCDJ
	var dst *net.UDPAddr
	if announce {
		vcdj, dst, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			if err := vcdj.Announce(ctx, c0, dst); err != nil {
				return err
			}
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return vcdj.KeepAlive(ctx, c0, dst)
		})
	}

	g.Go(func() error {
		n := 0
		for {
			select {
			case <-ctx.Done():
				fmt.Fprintf(os.Stderr, "captured %d packets to %s\n", n, out)
				return nil
			case p := <-pkts:
				if err := enc.Encode(p); err != nil {
					return errors.WithStack(err)
				}
				n++
			}
		}
	})

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// online discovers a real device to learn the local interface, then returns a
// virtual CDJ and the broadcast destination for its announce packets.
func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.UDPAddr, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})

	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}

		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{
			Device: device,
			IP:     ipNet.IP,
			MAC:    iface.HardwareAddr,
			Name:   name,
		}, &net.UDPAddr{IP: link.Broadcast(ipNet), Port: 50000}, nil
	}
}

func listen(ctx context.Context, c *net.UDPConn, port int, pkts chan<- Packet) error {
	go func() {
		<-ctx.Done()
		c.SetReadDeadline(time.Now())
	}()

	buf := make([]byte, 1500)
	for {
		n, addr, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		select {
		case pkts <- Packet{
			Data: fmt.Sprintf("%x", buf[:n]),
			Len:  n,
			Port: port,
			Src:  addr.String(),
			TS:   time.Now().UnixMilli(),
		}:
		case <-ctx.Done():
			return nil
		}
	}
}


The file /Users/noah/conductor/workspaces/deadca7/san-diego/link/cmd/capture/main.go has been updated successfully. (file state is current in your context — no need to Read it back)

// Command capture records raw Pro DJ Link UDP datagrams to a JSONL fixture.
//
// It listens on the DJ Link discovery, beat, and status ports and writes one
// Packet per line so tests can replay real traffic without hardware. With
// -announce it first brings a virtual CDJ online (the full startup handshake)
// so real players send status, metadata, and database traffic to it.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

// Packet is one recorded UDP datagram.
type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	announce := flag.Bool("announce", false, "bring a virtual CDJ online while capturing")
	device := flag.Int("device", 5, "virtual CDJ device number")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "capture.jsonl", "output JSONL fixture path")
	secs := flag.Int("secs", 8, "capture duration in seconds")
	flag.Parse()

	if err := run(*out, time.Duration(*secs)*time.Second, *announce, byte(*device), *name); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

func run(out string, d time.Duration, announce bool, device byte, name string) error {
	f, err := os.Create(out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, d)
	defer cancel()

	c0, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50000})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}

	c1, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50001})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c1.Close()

	c2, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50002})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c2.Close()

	var vcdj *link.VCDJ
	var dst *net.UDPAddr
	if announce {
		vcdj, dst, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			if err := vcdj.Announce(ctx, c0, dst); err != nil {
				return err
			}
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return vcdj.KeepAlive(ctx, c0, dst)
		})
	}

	g.Go(func() error {
		n := 0
		for {
			select {
			case <-ctx.Done():
				fmt.Fprintf(os.Stderr, "captured %d packets to %s\n", n, out)
				return nil
			case p := <-pkts:
				if err := enc.Encode(p); err != nil {
					return errors.WithStack(err)
				}
				n++
			}
		}
	})

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// online discovers a real device to learn the local interface, then returns a
// virtual CDJ and the broadcast destination for its announce packets.
func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.UDPAddr, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})

	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}

		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{
			Device: device,
			IP:     ipNet.IP,
			MAC:    iface.HardwareAddr,
			Name:   name,
		}, &net.UDPAddr{IP: link.Broadcast(ipNet), Port: 50000}, nil
	}
}

func listen(ctx context.Context, c *net.UDPConn, port int, pkts chan<- Packet) error {
	go func() {
		<-ctx.Done()
		c.SetReadDeadline(time.Now())
	}()

	buf := make([]byte, 1500)
	for {
		n, addr, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		select {
		case pkts <- Packet{
			Data: fmt.Sprintf("%x", buf[:n]),
			Len:  n,
			Port: port,
			Src:  addr.String(),
			TS:   time.Now().UnixMilli(),
		}:
		case <-ctx.Done():
			return nil
		}
	}
}

-// It passively listens on the DJ Link discovery, beat, and status ports and

-// writes one Packet per line so tests can replay real CDJ traffic without

-// hardware. Active flows (status, metadata, NFS) that require announcing a

-// virtual CDJ are not yet captured; see link/TODO.md.

+// It listens on the DJ Link discovery, beat, and status ports and writes one

+// Packet per line so tests can replay real traffic without hardware. With

+// -announce it first brings a virtual CDJ online (the full startup handshake)

+// so real players send status, metadata, and database traffic to it.

+  announce := flag.Bool("announce", false, "bring a virtual CDJ online while capturing")

+  device := flag.Int("device", 5, "virtual CDJ device number")

+  name := flag.String("name", "deadca7", "virtual CDJ device name")

-  if err := run(*out, time.Duration(*secs)*time.Second); err != nil {

+  if err := run(*out, time.Duration(*secs)*time.Second, *announce, byte(*device), *name); err != nil {

-func run(out string, d time.Duration) error {

+func run(out string, d time.Duration, announce bool, device byte, name string) error {

+  c0, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50000})

+  if err := link.SetBroadcast(c0); err != nil {

+  c1, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50001})

+  c2, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50002})

+    vcdj, dst, err = online(c0, device, name)

-    g.Go(func() error { return listen(ctx, port, pkts) })

+  g.Go(func() error { return listen(ctx, c0, 50000, pkts) })

+  g.Go(func() error { return listen(ctx, c1, 50001, pkts) })

+  g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

+      if err := vcdj.Announce(ctx, c0, dst); err != nil {

+      fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)

+      return vcdj.KeepAlive(ctx, c0, dst)

-func listen(ctx context.Context, port int, pkts chan<- Packet) error {

-  c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})

+// online discovers a real device to learn the local interface, then returns a

+// virtual CDJ and the broadcast destination for its announce packets.

+func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.UDPAddr, error) {

+  c0.SetReadDeadline(time.Now().Add(5 * time.Second))

+  defer c0.SetReadDeadline(time.Time{})

+      return nil, nil, errors.WithStack(err)

+    k, err := link.DecodeKeepAlive(buf[:n])

+    if err != nil || k.Device == device {

+    iface, ipNet, err := link.Iface(k.IP)

+    }, &net.UDPAddr{IP: link.Broadcast(ipNet), Port: 50000}, nil

+func listen(ctx context.Context, c *net.UDPConn, port int, pkts chan<- Packet) error {

// Command capture records raw Pro DJ Link UDP datagrams to a JSONL fixture.
//
// It passively listens on the DJ Link discovery, beat, and status ports and
// writes one Packet per line so tests can replay real CDJ traffic without
// hardware. Active flows (status, metadata, NFS) that require announcing a
// virtual CDJ are not yet captured; see link/TODO.md.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"time"

	"github.com/cockroachdb/errors"
	"golang.org/x/sync/errgroup"
)

// Packet is one recorded UDP datagram.
type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	out := flag.String("out", "capture.jsonl", "output JSONL fixture path")
	secs := flag.Int("secs", 8, "capture duration in seconds")
	flag.Parse()

	if err := run(*out, time.Duration(*secs)*time.Second); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

func run(out string, d time.Duration) error {
	ports := []int{50000, 50001, 50002}

	f, err := os.Create(out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, d)
	defer cancel()

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)

	g, ctx := errgroup.WithContext(ctx)
	for _, port := range ports {
		g.Go(func() error { return listen(ctx, port, pkts) })
	}

	g.Go(func() error {
		n := 0
		for {
			select {
			case <-ctx.Done():
				fmt.Fprintf(os.Stderr, "captured %d packets to %s\n", n, out)
				return nil
			case p := <-pkts:
				if err := enc.Encode(p); err != nil {
					return errors.WithStack(err)
				}
				n++
			}
		}
	})

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

func listen(ctx context.Context, port int, pkts chan<- Packet) error {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c.Close()

	go func() {
		<-ctx.Done()
		c.Close()
	}()

	buf := make([]byte, 1500)
	for {
		n, addr, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		select {
		case pkts <- Packet{
			Data: fmt.Sprintf("%x", buf[:n]),
			Len:  n,
			Port: port,
			Src:  addr.String(),
			TS:   time.Now().UnixMilli(),
		}:
		case <-ctx.Done():
			return nil
		}
	}
}


initialization failed: packages.Load error: err: exit status 1: stderr: go: go.work requires go >= 1.26.4 (running go 1.26.0)


Bash

cd /Users/noah/conductor/workspaces/deadca7/san-diego/link && go mod tidy >/dev/null 2>&1; go vet ./... 2>&1 | tail; echo "---TEST---"; go test ./... 2>&1 | tail

Tidy, vet, test the link module


---TEST---
ok  	github.com/nzoschke/deadca7/link	0.216s
?   	github.com/nzoschke/deadca7/link/cmd/capture	[no test files]


Everything's compiling and tests are passing, so now I need to run the actual capture with announce enabled to verify we're getting the 0x0a status packets from the CDJ-3000 and confirm the handshake is working properly. I'll run it for about 12 seconds to see if it holds up.


Builds and tests pass. Now the real validation — run the virtual CDJ with the full handshake and see whether the CDJ-3000 starts sending `0x0a` status (which never appeared passively):


go run ./cmd/capture -announce -device 5 -out testdata/status.jsonl -secs 12 2>&1; echo "---SUMMARY---"; python3 -c '
import json,collections
seen=collections.OrderedDict()
for line in open("testdata/status.jsonl"):
    p=json.loads(line); d=bytes.fromhex(p["data"])
    magic=d[:10]==bytes.fromhex("5173707431576d4a4f4c")
    typ=d[0x0a] if magic and len(d)>0x0a else None
    key=(p["src"].split(":")[0],p["port"],typ)
    e=seen.setdefault(key,[0,set()]); e[0]+=1; e[1].add(p["len"])
for (src,port,typ),(cnt,lens) in sorted(seen.items()):
    t=f"0x{typ:02x}" if typ is not None else "??"
    print(f"{src:15} port={port} type={t} count={cnt:4} lens={sorted(lens)}")
'

Capture with virtual CDJ announce


online as "deadca7" device 5
captured 403 packets to testdata/status.jsonl
---SUMMARY---
192.168.0.160   port=50000 type=0x00 count=   3 lens=[44]
192.168.0.160   port=50000 type=0x02 count=   3 lens=[50]
192.168.0.160   port=50000 type=0x04 count=   3 lens=[38]
192.168.0.160   port=50000 type=0x06 count=   6 lens=[54]
192.168.0.160   port=50000 type=0x0a count=   3 lens=[38]
192.168.0.192   port=50000 type=0x06 count=   4 lens=[54]
192.168.0.192   port=50001 type=0x0b count= 273 lens=[60]
192.168.0.192   port=50002 type=0x0a count=  49 lens=[1116]
192.168.0.192   port=50002 type=0x40 count=   5 lens=[38]
192.168.0.193   port=50000 type=0x05 count=   3 lens=[38]
192.168.0.193   port=50000 type=0x06 count=   6 lens=[54]
192.168.0.193   port=50001 type=0x03 count=  39 lens=[53]
192.168.0.193   port=50001 type=0x04 count=   6 lens=[100]


 The handshake worked—the CDJ-3000 is now actively transmitting status packets on port 50002, which were previously unavailable. I'm seeing the full CDJ status data along with high-frequency position and sync information on a separate port, confirming the virtual CDJ announcement successfully unlocked the device's communication.

Our handshake sequence is also visible in the traffic: the claim packets, keepalives, and the mixer's response packets all match the spec exactly. This validates the four-stage handshake approach and proves it's more reliable than the shortcut method. Now I want to run this again to confirm the 0x0a status packets appear consistently—that's the main concern about stability—and then I can start decoding key fields from the status packet like player number, BPM, play state, and track ID to verify the fixture is capturing the right data.


The spec-compliant handshake **works and unlocked the active flow** that the vizlab shortcut couldn't reliably get:

- Our virtual CDJ sent the **exact spec sequence** from `192.168.0.160`: `0x0a` hello ×3 → `0x00` claim1 ×3 → `0x02` claim2 ×3 → `0x04` claim3 ×3 → `0x06` keep-alives.
- **The CDJ-3000 responded with `0x0a` status — 49 packets, len 1116** (never present in any passive capture), plus a high-rate `0x0b` position stream on 50001.
- The **DJM-V10 reacted to the claim** with `0x05` packets — i.e. the mixer acknowledged us, exactly the dialog the old code was missing.

Let me confirm it's *stable* (the actual complaint) by running it again — instability would show as missing/intermittent `0x0a`:


for run in 1 2; do go run ./cmd/capture -announce -device 5 -out /tmp/s$run.jsonl -secs 8 2>/dev/null; python3 -c "
import json
c=sum(1 for l in open('/tmp/s$run.jsonl') if json.loads(l)['port']==50002 and bytes.fromhex(json.loads(l)['data'])[0x0a]==0x0a)
print(f'run $run: 0x0a status packets = {c}')
"; done

Two more runs to verify stability


run 1: 0x0a status packets = 24
run 2: 0x0a status packets = 32


 The status packet is arriving consistently on every run, which is solid. I should update the TODO list to mark the handshake as complete and note that status capture is working, then document the new packet types I've discovered—0x0b for position data on port 50001, 0x05 for mixer assignment on 50000, and 0x04 for mixer on 50001. I can add a basic test to verify the status fixture contains the presence packets, and the next step is tackling the status decode logic.

Now I need to review the findings against the spec and implementation as the user requested. Let me update the TODO file to reflect the discovery progress—marking the handshake as captured and tested, adding the new packet types to the status section, and noting what's been confirmed in the reliability section. I'll check the current TODO structure to make targeted edits.


Stable — status arrives every run (24, 32 packets). The instability is fixed. Let me save the fixture and update `TODO.md`.


19	read together with the port.
20
21	### Discovery / keep-alive — port 50000
22	- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
23	      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
24	- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode
25	- [ ] full announce/channel-negotiation handshake (claim device # 1-4)
26	- [ ] virtual CDJ Announce + KeepAlive sender (prereq for all active flows)
27
28	### Beat sync — port 50001
29	- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
30	      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
31	- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
32	- [ ] mixer type 0x03 (~3/s from DJM-V10, name at 0x0b not 0x0c) — identify & decode
33	- [ ] capture across a tempo change and a loaded vs. stopped deck
34
35	### Status — port 50002
36	- NOTE: passively only the CDJ's low-rate type 0x40 ("NXS-GW", ~0.5/s) appears.
37	  Full 0x0a status is NOT broadcast — it requires announcing a virtual CDJ first.
38	- [ ] CDJ status (type 0x0a): play state, BPM, pitch, on-air, master, sync
39	- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
40	- [ ] mixer status (DJM) if present
41	- [ ] CDJ type 0x40 on 50002 — identify & decode
42	- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle
43
44	## crate-digger — rekordbox database (over NFS)
45
46	Reached only after the virtual CDJ announces and a deck reports a loaded track.
47
48	- [ ] RPC portmap (port 111) — GETPORT for mount + nfs
49	- [ ] MOUNT — mount the media slot, obtain root file handle
50	- [ ] NFS — LOOKUP + READ `PIONEER/rekordbox/export.pdb`
51	- [ ] PDB parse: tables (tracks, artists, albums, artwork, playlists)
52	- [ ] ANLZ parse: `.DAT`/`.EXT` (beat grid, cue points, waveforms)
53	- [ ] metadata-by-track-ID resolved against captured PDB
54

read together with the port.

### Discovery / keep-alive — port 50000
- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode
- [ ] full announce/channel-negotiation handshake (claim device # 1-4)
- [ ] virtual CDJ Announce + KeepAlive sender (prereq for all active flows)

### Beat sync — port 50001
- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
- [ ] mixer type 0x03 (~3/s from DJM-V10, name at 0x0b not 0x0c) — identify & decode
- [ ] capture across a tempo change and a loaded vs. stopped deck

### Status — port 50002
- NOTE: passively only the CDJ's low-rate type 0x40 ("NXS-GW", ~0.5/s) appears.
  Full 0x0a status is NOT broadcast — it requires announcing a virtual CDJ first.
- [ ] CDJ status (type 0x0a): play state, BPM, pitch, on-air, master, sync
- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
- [ ] mixer status (DJM) if present
- [ ] CDJ type 0x40 on 50002 — identify & decode
- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle

## crate-digger — rekordbox database (over NFS)

Reached only after the virtual CDJ announces and a deck reports a loaded track.

- [ ] RPC portmap (port 111) — GETPORT for mount + nfs
- [ ] MOUNT — mount the media slot, obtain root file handle
- [ ] NFS — LOOKUP + READ `PIONEER/rekordbox/export.pdb`
- [ ] PDB parse: tables (tracks, artists, albums, artwork, playlists)
- [ ] ANLZ parse: `.DAT`/`.EXT` (beat grid, cue points, waveforms)
- [ ] metadata-by-track-ID resolved against captured PDB


Confirm status fixture saved


403 testdata/status.jsonl


Edit

### Discovery / keep-alive — port 50000
- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode
- [ ] full announce/channel-negotiation handshake (claim device # 1-4)
- [ ] virtual CDJ Announce + KeepAlive sender (prereq for all active flows)

### Beat sync — port 50001
- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
- [ ] mixer type 0x03 (~3/s from DJM-V10, name at 0x0b not 0x0c) — identify & decode
- [ ] capture across a tempo change and a loaded vs. stopped deck

### Status — port 50002
- NOTE: passively only the CDJ's low-rate type 0x40 ("NXS-GW", ~0.5/s) appears.
  Full 0x0a status is NOT broadcast — it requires announcing a virtual CDJ first.
- [ ] CDJ status (type 0x0a): play state, BPM, pitch, on-air, master, sync
- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
- [ ] mixer status (DJM) if present
- [ ] CDJ type 0x40 on 50002 — identify & decode
- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle

### Discovery / startup — port 50000
- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
- [x] virtual CDJ four-stage startup handshake (announce.go): hello (0x0a) ×3,
      claim 0x00/0x02/0x04 ×3 at 300ms, then keep-alive (0x06) at 1500ms.
      Verified the CDJ-3000 then streams 0x0a status — `testdata/status.jsonl`.
- [ ] device-number conflict handling: react to a defense (0x08/0x29) and a
      mixer assignment (DJM replies 0x05 to our claim) instead of fixed number
- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode

### Beat sync — port 50001
- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
- [ ] CDJ type 0x0b position stream (~30/s, len 60) — identify & decode
- [ ] mixer type 0x03 (~3/s) and type 0x04 (len 100) from DJM-V10 — decode
- [ ] capture across a tempo change and a loaded vs. stopped deck

### Status — port 50002 (needs virtual CDJ online)
- [x] captured CDJ 0x0a status (len 1116) — `testdata/status.jsonl`
- [ ] decode 0x0a: play state, BPM, pitch, on-air, master, sync
- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
- [ ] CDJ type 0x40 on 50002 — identify & decode
- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle


The file /Users/noah/conductor/workspaces/deadca7/san-diego/link/TODO.md has been updated successfully. (file state is current in your context — no need to Read it back)

### Discovery / startup — port 50000
- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
- [x] virtual CDJ four-stage startup handshake (announce.go): hello (0x0a) ×3,
      claim 0x00/0x02/0x04 ×3 at 300ms, then keep-alive (0x06) at 1500ms.
      Verified the CDJ-3000 then streams 0x0a status — `testdata/status.jsonl`.
- [ ] device-number conflict handling: react to a defense (0x08/0x29) and a
      mixer assignment (DJM replies 0x05 to our claim) instead of fixed number
- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode

### Beat sync — port 50001
- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
- [ ] CDJ type 0x0b position stream (~30/s, len 60) — identify & decode
- [ ] mixer type 0x03 (~3/s) and type 0x04 (len 100) from DJM-V10 — decode
- [ ] capture across a tempo change and a loaded vs. stopped deck

### Status — port 50002 (needs virtual CDJ online)
- [x] captured CDJ 0x0a status (len 1116) — `testdata/status.jsonl`
- [ ] decode 0x0a: play state, BPM, pitch, on-air, master, sync
- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
- [ ] CDJ type 0x40 on 50002 — identify & decode
- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle

# link TODO

Go port of Pioneer Pro DJ Link, built capture-first: document each packet,
capture a real fixture from the CDJ-3000, then decode it under test. No test
should require live hardware — `cmd/capture` records fixtures into `testdata/`.

## Setup

- CDJ-3000 (device 3) at `192.168.0.192`, DJM-V10 mixer (device 33) at
  `192.168.0.193`, this Mac at `192.168.0.160` (en0), one `/24`.
- Reference SD card loaded in the CDJ for the database (crate-digger) flows.
- Passive capture: `go run ./cmd/capture -out testdata/<name>.jsonl -secs 8`.
- Active flows need a virtual CDJ that announces first — see Announce below.
  RPC/NFS uses dynamic ports, so capture those with `tcpdump` alongside.

## beat-link — live network protocol (UDP)

Each packet starts with Magic `51 73 70 74 31 57 6d 4a 4f 4c` + a type byte
read together with the port.

### Discovery / keep-alive — port 50000
- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode
- [ ] full announce/channel-negotiation handshake (claim device # 1-4)
- [ ] virtual CDJ Announce + KeepAlive sender (prereq for all active flows)

### Beat sync — port 50001
- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
- [ ] mixer type 0x03 (~3/s from DJM-V10, name at 0x0b not 0x0c) — identify & decode
- [ ] capture across a tempo change and a loaded vs. stopped deck

### Status — port 50002
- NOTE: passively only the CDJ's low-rate type 0x40 ("NXS-GW", ~0.5/s) appears.
  Full 0x0a status is NOT broadcast — it requires announcing a virtual CDJ first.
- [ ] CDJ status (type 0x0a): play state, BPM, pitch, on-air, master, sync
- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
- [ ] mixer status (DJM) if present
- [ ] CDJ type 0x40 on 50002 — identify & decode
- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle

## crate-digger — rekordbox database (over NFS)

Reached only after the virtual CDJ announces and a deck reports a loaded track.

- [ ] RPC portmap (port 111) — GETPORT for mount + nfs
- [ ] MOUNT — mount the media slot, obtain root file handle
- [ ] NFS — LOOKUP + READ `PIONEER/rekordbox/export.pdb`
- [ ] PDB parse: tables (tracks, artists, albums, artwork, playlists)
- [ ] ANLZ parse: `.DAT`/`.EXT` (beat grid, cue points, waveforms)
- [ ] metadata-by-track-ID resolved against captured PDB

## Reliability (port the vizlab code without its known bugs)

- [ ] replace `time.Sleep(5s)` FIXME in `Start()` with discovery/mount readiness
- [ ] make the media query deterministic (vizlab's was "unreliable")
- [ ] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`
- [ ] live integration suite gated behind `testing.Short()` / env, separate from
      the hermetic fixture-replay tests

-### Discovery / keep-alive — port 50000

+- [x] virtual CDJ four-stage startup handshake (announce.go): hello (0x0a) ×3,

+      claim 0x00/0x02/0x04 ×3 at 300ms, then keep-alive (0x06) at 1500ms.

+      Verified the CDJ-3000 then streams 0x0a status — `testdata/status.jsonl`.

+- [ ] device-number conflict handling: react to a defense (0x08/0x29) and a

+      mixer assignment (DJM replies 0x05 to our claim) instead of fixed number

-- [ ] full announce/channel-negotiation handshake (claim device # 1-4)

-- [ ] virtual CDJ Announce + KeepAlive sender (prereq for all active flows)

-- [ ] mixer type 0x03 (~3/s from DJM-V10, name at 0x0b not 0x0c) — identify & decode

+- [ ] CDJ type 0x0b position stream (~30/s, len 60) — identify & decode

+- [ ] mixer type 0x03 (~3/s) and type 0x04 (len 100) from DJM-V10 — decode

-- NOTE: passively only the CDJ's low-rate type 0x40 ("NXS-GW", ~0.5/s) appears.

-  Full 0x0a status is NOT broadcast — it requires announcing a virtual CDJ first.

-- [ ] CDJ status (type 0x0a): play state, BPM, pitch, on-air, master, sync

+### Status — port 50002 (needs virtual CDJ online)

+- [x] captured CDJ 0x0a status (len 1116) — `testdata/status.jsonl`

+- [ ] decode 0x0a: play state, BPM, pitch, on-air, master, sync


- [ ] replace `time.Sleep(5s)` FIXME in `Start()` with discovery/mount readiness
- [ ] make the media query deterministic (vizlab's was "unreliable")
- [ ] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`
- [ ] live integration suite gated behind `testing.Short()` / env, separate from
      the hermetic fixture-replay tests

- [x] proper startup handshake instead of vizlab blasting only keep-alives —
      this was the main instability: real players need hello + claim first
- [x] SetBroadcast (SO_BROADCAST) so subnet-broadcast sends don't silently fail
- [ ] replace `time.Sleep(5s)` FIXME in `Start()` with discovery/mount readiness
- [ ] make the media query deterministic (vizlab's was "unreliable")
- [x] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`
- [ ] live integration suite gated behind `testing.Short()` / env, separate from
      the hermetic fixture-replay tests


- [x] proper startup handshake instead of vizlab blasting only keep-alives —
      this was the main instability: real players need hello + claim first
- [x] SetBroadcast (SO_BROADCAST) so subnet-broadcast sends don't silently fail
- [ ] replace `time.Sleep(5s)` FIXME in `Start()` with discovery/mount readiness
- [ ] make the media query deterministic (vizlab's was "unreliable")
- [x] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`
- [ ] live integration suite gated behind `testing.Short()` / env, separate from
      the hermetic fixture-replay tests

# link TODO

Go port of Pioneer Pro DJ Link, built capture-first: document each packet,
capture a real fixture from the CDJ-3000, then decode it under test. No test
should require live hardware — `cmd/capture` records fixtures into `testdata/`.

## Setup

- CDJ-3000 (device 3) at `192.168.0.192`, DJM-V10 mixer (device 33) at
  `192.168.0.193`, this Mac at `192.168.0.160` (en0), one `/24`.
- Reference SD card loaded in the CDJ for the database (crate-digger) flows.
- Passive capture: `go run ./cmd/capture -out testdata/<name>.jsonl -secs 8`.
- Active flows need a virtual CDJ that announces first — see Announce below.
  RPC/NFS uses dynamic ports, so capture those with `tcpdump` alongside.

## beat-link — live network protocol (UDP)

Each packet starts with Magic `51 73 70 74 31 57 6d 4a 4f 4c` + a type byte
read together with the port.

### Discovery / startup — port 50000
- [x] keep-alive (type 0x06): name, device #, IP, MAC — `testdata/devices.jsonl`
      covers CDJ-3000 (device 3) and DJM-V10 (device 33); name at 0x0c
- [x] virtual CDJ four-stage startup handshake (announce.go): hello (0x0a) ×3,
      claim 0x00/0x02/0x04 ×3 at 300ms, then keep-alive (0x06) at 1500ms.
      Verified the CDJ-3000 then streams 0x0a status — `testdata/status.jsonl`.
- [ ] device-number conflict handling: react to a defense (0x08/0x29) and a
      mixer assignment (DJM replies 0x05 to our claim) instead of fixed number
- [ ] mixer/gateway packet seen on 50002 (type 0x40, "NXS-GW") — identify & decode

### Beat sync — port 50001
- [x] beat packet (type 0x28): BPM, beat-in-bar, pitch — `testdata/beats.jsonl`
      (name@0x0b; bpm@0x5a centi; beat@0x5c; pitch@0x55 3-byte, 0x100000=100%)
- [ ] next/prev beat-timing distances (six 4-byte BE values from ~0x24)
- [ ] CDJ type 0x0b position stream (~30/s, len 60) — identify & decode
- [ ] mixer type 0x03 (~3/s) and type 0x04 (len 100) from DJM-V10 — decode
- [ ] capture across a tempo change and a loaded vs. stopped deck

### Status — port 50002 (needs virtual CDJ online)
- [x] captured CDJ 0x0a status (len 1116) — `testdata/status.jsonl`
- [ ] decode 0x0a: play state, BPM, pitch, on-air, master, sync
- [ ] loaded-track fields: track ID, source player, source slot (SD/USB)
- [ ] CDJ type 0x40 on 50002 — identify & decode
- [ ] capture scenarios: idle, track loaded, playing, two-deck, on-air toggle

## crate-digger — rekordbox database (over NFS)

Reached only after the virtual CDJ announces and a deck reports a loaded track.

- [ ] RPC portmap (port 111) — GETPORT for mount + nfs
- [ ] MOUNT — mount the media slot, obtain root file handle
- [ ] NFS — LOOKUP + READ `PIONEER/rekordbox/export.pdb`
- [ ] PDB parse: tables (tracks, artists, albums, artwork, playlists)
- [ ] ANLZ parse: `.DAT`/`.EXT` (beat grid, cue points, waveforms)
- [ ] metadata-by-track-ID resolved against captured PDB

## Reliability (port the vizlab code without its known bugs)

- [ ] replace `time.Sleep(5s)` FIXME in `Start()` with discovery/mount readiness
- [ ] make the media query deterministic (vizlab's was "unreliable")
- [ ] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`
- [ ] live integration suite gated behind `testing.Short()` / env, separate from
      the hermetic fixture-replay tests

+- [x] proper startup handshake instead of vizlab blasting only keep-alives —

+      this was the main instability: real players need hello + claim first

+- [x] SetBroadcast (SO_BROADCAST) so subnet-broadcast sends don't silently fail

-- [ ] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`

+- [x] swap `github.com/pkg/errors` → `github.com/cockroachdb/errors`


Now I'm pulling together the final review with the root cause analysis, other issues I found, what I implemented to fix them, and the validation results to show the user what's next.


Done. Here's the review you asked for, plus the fix I implemented and validated.

## Review: why the vizlab virtual CDJ was unstable

I checked the [Deep-Symmetry startup spec](https://djl-analysis.deepsymmetry.org/djl-analysis/startup.html) against the vizlab `announce.go`. The spec requires a **four-stage startup handshake** before keep-alives:

| Stage | Type | Count | Interval |
|---|---|---|---|
| hello | `0x0a` | 3 | 300ms |
| claim stage 1 | `0x00` | 3 | 300ms |
| claim stage 2 | `0x02` | 3 | 300ms |
| claim stage 3 | `0x04` | 3 | 300ms |
| keep-alive | `0x06` | ∞ | 1500ms |

**The vizlab code skips stages 1–4 entirely** — it sends *one* keep-alive in `Announce()` then a 1.5s ticker. That's the root cause:

1. **No handshake** → real players (especially with a DJM mixer present) never properly register the virtual CDJ, so status delivery is flaky/absent. This matches what we saw: zero `0x0a` status in any passive capture.
2. **No `SO_BROADCAST`** → on macOS, sending to the subnet broadcast can silently fail. Go doesn't set it by default; vizlab never did.
3. **Fixed device number, no conflict handling** → it ignores defense packets (`0x08`/`0x29`) and mixer assignments, so a number collision leaves it half-online.
4. **Timing races** → single keep-alive + separate ticker + the `time.Sleep(5s)` FIXME instead of waiting on a real "online" signal.

Plus the general-implementation nits from CLAUDE.md: `pkg/errors` instead of `cockroachdb/errors`.

## What I built (`link/announce.go`, `net.go`)

A spec-compliant `VCDJ` using **beat-link's exact byte templates**: `Announce()` runs hello + the three claim stages at 300ms, returns when "online", then `KeepAlive()` holds the number at 1500ms (sends immediately — no initial gap). Added `SetBroadcast()`. Wired an `-announce` flag into the capture tool. New hermetic test `TestVCDJPackets` validates every packet's length, type, and patched fields (round-tripping keep-alive through `DecodeKeepAlive`) — none of this was tested before.

## Validation against the CDJ-3000

The handshake **works and is stable**. With `-announce`, the CDJ-3000 immediately began streaming the rich **`0x0a` status (len 1116)** that never appeared passively, and the **DJM-V10 acknowledged our claim with `0x05`** packets. Three consecutive runs all got status (49, 24, 32 packets) — no intermittency. Saved as `testdata/status.jsonl`.

`go vet` + `go test` clean.

## Still open (tracked in TODO.md)
- **Conflict robustness**: react to a defense (`0x08`) and to the mixer's `0x05` assignment, rather than a fixed number — the remaining correctness gap before this is production-stable.
- **Decode `0x0a` status**: play state, BPM/pitch, on-air, and the **loaded track ID + slot** — that's the gateway into the crate-digger NFS/PDB side.
- New packets the handshake surfaced: CDJ `0x0b` position stream (50001), mixer `0x04` (50001).

Want me to harden the device-number negotiation (defense + mixer-assign) next, or start decoding the `0x0a` status packet now that we have a solid fixture?


yes do the device number negotiation thing while we're here


Now I need to implement the device-number negotiation logic—handling defense packets when they arrive, processing mixer assignments from the DJM, and self-assigning a device number when it's zero. I'm looking at beat-link's approach: the `selfAssignDeviceNumber()` method picks an available number, defense packets (0x08/0x29) trigger a claim rejection flag, and the mixer sends assignment packets that allocate numbers to devices.


Let me study beat-link's negotiation logic — self-assign, defense detection, and mixer assignment — before implementing.


Bash

Read selfAssignDeviceNumber


* The number of milliseconds for which the {@link DeviceFinder} needs to have been watching the network in order
     * for us to be confident we can choose a device number that will not conflict.
     */
    private static final long SELF_ASSIGNMENT_WATCH_PERIOD = 4000;

    /**
     *
Try to choose a device number, which we have not seen on the network. If we have already tried one, it must
     * have been defended, so increment to the next one we can try (we stop at 15). If we have not yet tried one,
     * pick the first appropriate one to try, honoring the value of {@link #useStandardPlayerNumber} to determine
     * if we start at 1 or 7. Set the number we are going to try next in {@link #claimingNumber}.

     *
     *
Even though in theory we should be able to rely on the protocol to tell us if we are claiming a
     * number that belongs to another player, it turns out the XDJ-XZ is buggy and tells us to go ahead and
     * use whatever we were claiming, and further fails to defend the device numbers that it is using. So
     * we still need to make sure the {@link DeviceFinder} has been running long enough to see all devices
     * so we can avoid trying to claim a number that some other device is already using.

     *
     * @return true if there was a number available for us to try claiming
     */
    private boolean selfAssignDeviceNumber() {
        long now = System.currentTimeMillis();
        final long started = DeviceFinder.getInstance().getFirstDeviceTime();
        while (now - started < SELF_ASSIGNMENT_WATCH_PERIOD) {
            final long sleepMillis = SELF_ASSIGNMENT_WATCH_PERIOD - (now - started);
            logger.debug("Not waited long enough for devices; sleeping for {}ms", sleepMillis);
            try {
                Thread.sleep(sleepMillis);  // Sleep until we hit the right time
            } catch (InterruptedException e) {
                logger.warn("Interrupted waiting to self-assign device number, giving up.");
                return false;
            }
            now = System.currentTimeMillis();
        }
        logger.debug("Self-assigning number {}ms after first device seen.", System.currentTimeMillis() - started);

        if (claimingNumber.get() == 0) {
            // We have not yet tried a number. If we are not supposed to use standard player numbers, make sure
            // the first one we try is 7 (to accommodate the CDJ-3000, which can use channels 5 and 6).
            if (!getUseStandardPlayerNumber()) {
                claimingNumber.set(6);
            }
        }

        // Record what numbers we have already seen, since there is no point trying one of them.
        Set<Integer> numbersUsed = new HashSet<>();
        for (DeviceAnnouncement device : DeviceFinder.getInstance().getCurrentDevices()) {
            numbersUsed.add(device.getDeviceNumber());
        }
        logger.debug("Player numbers seen on network: {}", numbersUsed);

        // Try next available player number less than mixers use.
        final int startingNumber = claimingNumber.get() + 1;
        for (int result = startingNumber; result < 16; result++) {
            if (!numbersUsed.contains(result)) {  // We found one that is not used, so we can use it
                claimingNumber.set(result);
                if (getUseStandardPlayerNumber() && (result > 4)) {
                    logger.warn("Unable to self-assign a standard player number, all are in use. Trying number {}.", result);
                }
                return true;
            }
        }
        logger.warn("Found no unused device numbers between {} and 15, giving up.", startingNumber);
        return false;
    }

    /**
     * Hold the network interfaces which match the address on which we found player traffic. Should only be one,
     * or we will likely receive duplicate packets, which will cause problematic behavior.
     */
    private List<NetworkInterface> matchingInterfaces = null;

    /**
     * Check the interfaces that match the address from which we are receiving DJ Link traffic. If there is more
     * than one value in this list, that is a problem because we will likely receive duplicate packets that will
     * play havoc with our understanding of player states.
     *
     * @return the list of network interfaces on which we might receive player packets
     * @throws IllegalStateException if we are not running
     */
    @API(status = API.Status.STABLE)
    public List<NetworkInterface> getMatchingInterfaces() {
        if (VirtualRekordbox.getInstance().isRunning()) {
            return VirtualRekordbox.getInstance().getMatchingInterfaces();
        }

        ensureRunning();
        return Collections.unmodifiableList(matchingInterfaces);
    }

    /**
     * Holds the interface address we chose to communicate with the DJ Link device we found during startup,
Shell cwd was reset to /Users/noah/conductor/workspaces/deadca7/san-diego


I need to search through the file to locate where claimRejected and mixerAssigned are being set on incoming packets.


Now the packet-handling side — where defense and mixer-assignment are detected:


cd /tmp && grep -n 'claimRejected\|mixerAssigned\|case 0x08\|case 0x02\|0x29\|getDeviceNumberToClaim\|requestNumberFromMixer\|assignmentRequestBytes\|handleDeviceClaim\|0x0a:\|0x04:\|0x08:' vcdj.java | head -40

Find defense/assignment packet handling


348:    private static final byte[] assignmentRequestBytes = {
361:            0x01, 0x02, 0x00, 0x29,  0x00, 0x00, 0x00, 0x00,   0x00
705:    private final AtomicBoolean claimRejected = new AtomicBoolean(false);
710:    private final AtomicInteger mixerAssigned = new AtomicInteger(0);
719:    private void requestNumberFromMixer(InetAddress mixerAddress) {
727:        Arrays.fill(assignmentRequestBytes, DEVICE_NAME_OFFSET, DEVICE_NAME_LENGTH, (byte)0);
728:        System.arraycopy(getDeviceName().getBytes(), 0, assignmentRequestBytes, DEVICE_NAME_OFFSET, getDeviceName().getBytes().length);
729:        System.arraycopy(matchedAddress.getAddress().getAddress(), 0, assignmentRequestBytes, 0x24, 4);
730:        System.arraycopy(keepAliveBytes, MAC_ADDRESS_OFFSET, assignmentRequestBytes, 0x28, 6);
732:        assignmentRequestBytes[0x31] = (keepAliveBytes[DEVICE_NUMBER_OFFSET] == 0)? (byte)1 : (byte)2;  // The auto-assign flag.
733:        assignmentRequestBytes[0x2f] = 1;  // The packet counter.
735:            DatagramPacket announcement = new DatagramPacket(assignmentRequestBytes, assignmentRequestBytes.length,
780:        claimRejected.set(false);
781:        mixerAssigned.set(0);
817:            for (int i = 1; i <= 3 && mixerAssigned.get() == 0; i++) {
831:                if (claimRejected.get()) {  // Some other player is defending the number we tried to claim.
849:            for (int i = 1; i <= 3 && mixerAssigned.get() == 0; i++) {
863:                if (claimRejected.get()) {  // Some other player is defending the number we tried to claim.
874:            final int assigned = mixerAssigned.getAndSet(0);
884:            for (int i = 1; i <= 3 && mixerAssigned.get() == 0; i++) {
898:                if (claimRejected.get()) {  // Some other player is defending the number we tried to claim.
912:        mixerAssigned.set(0);
2006:                    if (number >= 0x29 && number < 0x30) { // We can see rekordbox mobile, this will work too.
2883:    private void handleDeviceClaimPacket(DatagramPacket packet, int deviceOffset) {
2904:            handleDeviceClaimPacket(packet, 0x2e);
2906:            handleDeviceClaimPacket(packet, 0x24);
2910:                requestNumberFromMixer(packet.getAddress());
2915:            mixerAssigned.set(packet.getData()[0x24]);
2916:            if (mixerAssigned.get() == 0) {
2919:                logger.info("Mixer at address {} told us to use device number {}", packet.getAddress().getHostAddress(), mixerAssigned.get());
2922:            mixerAssigned.set(claimingNumber.get());
2930:                claimRejected.set(true);
Shell cwd was reset to /Users/noah/conductor/workspaces/deadca7/san-diego


cd /tmp && sed -n '2883,2935p' vcdj.java; echo "=== packet dispatch around 2900 ==="; sed -n '2890,2912p' vcdj.java

Read claim/assignment/defense handlers


    private void handleDeviceClaimPacket(DatagramPacket packet, int deviceOffset) {
        if (packet.getData().length < deviceOffset + 1) {
            logger.warn("Ignoring too-short device claim packet.");
            return;
        }
        if (isRunning() && getDeviceNumber() == packet.getData()[deviceOffset]) {
            defendDeviceNumber(packet.getAddress());
        }
    }

    /**
     * The {@link DeviceFinder} delegates packets it doesn't know how to deal with to us using this method, because
     * they relate to claiming or defending device numbers, which is our responsibility.
     *
     * @param kind the kind of packet that was received
     * @param packet the actual bytes of the packet
     */
    void handleSpecialAnnouncementPacket(Util.PacketType kind, DatagramPacket packet) {
        if (kind == Util.PacketType.DEVICE_NUMBER_STAGE_1) {
            logger.debug("Received device number claim stage 1 packet.");
        } else if (kind == Util.PacketType.DEVICE_NUMBER_STAGE_2) {
            handleDeviceClaimPacket(packet, 0x2e);
        } else if (kind == Util.PacketType.DEVICE_NUMBER_STAGE_3) {
            handleDeviceClaimPacket(packet, 0x24);
        } else if (kind == Util.PacketType.DEVICE_NUMBER_WILL_ASSIGN) {
            logger.debug("The mixer at address {} wants to assign us a specific device number.", packet.getAddress().getHostAddress());
            if (claimingNumber.get() != 0) {
                requestNumberFromMixer(packet.getAddress());
            } else {
                logger.warn("Ignoring mixer device number assignment offer; we are not claiming a device number!");
            }
        } else if (kind == Util.PacketType.DEVICE_NUMBER_ASSIGN) {
            mixerAssigned.set(packet.getData()[0x24]);
            if (mixerAssigned.get() == 0) {
                logger.debug("Mixer at address {} told us to use any device.", packet.getAddress().getHostAddress());
            } else {
                logger.info("Mixer at address {} told us to use device number {}", packet.getAddress().getHostAddress(), mixerAssigned.get());
            }
        } else if (kind == Util.PacketType.DEVICE_NUMBER_ASSIGNMENT_FINISHED) {
            mixerAssigned.set(claimingNumber.get());
            logger.info("Mixer confirmed device assignment.");
        } else if (kind == Util.PacketType.DEVICE_NUMBER_IN_USE) {
            final int defendedDevice = packet.getData()[0x24];
            if (defendedDevice == 0) {
                logger.warn("Ignoring unexplained attempt to defend device 0.");
            } else if (defendedDevice == claimingNumber.get()) {
                logger.warn("Another device is defending device number {}, so we can't use it.", defendedDevice);
                claimRejected.set(true);
            } else if (isRunning()) {
                if (defendedDevice == getDeviceNumber()) {
                    logger.warn("Another device has claimed it owns our device number, shutting down.");
                    stop();
                } else if (defendedDevice == 25) {
=== packet dispatch around 2900 ===
        }
    }

    /**
     * The {@link DeviceFinder} delegates packets it doesn't know how to deal with to us using this method, because
     * they relate to claiming or defending device numbers, which is our responsibility.
     *
     * @param kind the kind of packet that was received
     * @param packet the actual bytes of the packet
     */
    void handleSpecialAnnouncementPacket(Util.PacketType kind, DatagramPacket packet) {
        if (kind == Util.PacketType.DEVICE_NUMBER_STAGE_1) {
            logger.debug("Received device number claim stage 1 packet.");
        } else if (kind == Util.PacketType.DEVICE_NUMBER_STAGE_2) {
            handleDeviceClaimPacket(packet, 0x2e);
        } else if (kind == Util.PacketType.DEVICE_NUMBER_STAGE_3) {
            handleDeviceClaimPacket(packet, 0x24);
        } else if (kind == Util.PacketType.DEVICE_NUMBER_WILL_ASSIGN) {
            logger.debug("The mixer at address {} wants to assign us a specific device number.", packet.getAddress().getHostAddress());
            if (claimingNumber.get() != 0) {
                requestNumberFromMixer(packet.getAddress());
            } else {
                logger.warn("Ignoring mixer device number assignment offer; we are not claiming a device number!");
Shell cwd was reset to /Users/noah/conductor/workspaces/deadca7/san-diego


I'm working through the packet structure for the mixer's negotiation flow on port 50000. Looking at the device assignment sequence, I need to map out the packet type bytes—the mixer sends a DEVICE_NUMBER_WILL_ASSIGN message (likely 0x05 based on the DJM's 38-byte packet) followed by a DEVICE_NUMBER_ASSIGN message that contains the actual device number.


Let me get the exact type-byte values for these negotiation packets from beat-link's `Util.PacketType`:


cd /tmp && gh api repos/Deep-Symmetry/beat-link/contents/src/main/java/org/deepsymmetry/beatlink/Util.java --jq '.content' 2>/dev/null | base64 -d | grep -nE 'DEVICE_NUMBER|DEVICE_HELLO|PacketType\(|0x0[0-9a-f]' | grep -iE 'DEVICE_NUMBER|HELLO|new byte|\(0x' | head -40

Find PacketType enum type bytes


64:        FADER_START_COMMAND(0x02, "Fader Start", BeatFinder.BEAT_PORT),
69:        CHANNELS_ON_AIR(0x03, "Channels On Air", BeatFinder.BEAT_PORT),
74:        MEDIA_QUERY(0x05, "Media Query", VirtualCdj.UPDATE_PORT),
79:        MEDIA_RESPONSE(0x06, "Media Response", VirtualCdj.UPDATE_PORT),
85:        DEVICE_HELLO(0x0a, "Device Hello", DeviceFinder.ANNOUNCEMENT_PORT),
99:        DEVICE_NUMBER_STAGE_1(0x00, "Device Number Claim Stage 1", DeviceFinder.ANNOUNCEMENT_PORT),
106:        DEVICE_NUMBER_WILL_ASSIGN(0x01, "Device Number Will Be Assigned", DeviceFinder.ANNOUNCEMENT_PORT),
111:        DEVICE_NUMBER_STAGE_2(0x02, "Device Number Claim Stage 2", DeviceFinder.ANNOUNCEMENT_PORT),
117:        DEVICE_NUMBER_ASSIGN(0x03, "Device Number Assignment", DeviceFinder.ANNOUNCEMENT_PORT),
123:        DEVICE_NUMBER_STAGE_3(0x04, "Device Number Claim Stage 3", DeviceFinder.ANNOUNCEMENT_PORT),
129:        DEVICE_NUMBER_ASSIGNMENT_FINISHED(0x05, "Device Number Assignment Finished", DeviceFinder.ANNOUNCEMENT_PORT),
134:        DEVICE_KEEP_ALIVE(0x06, "Device Keep-Alive", DeviceFinder.ANNOUNCEMENT_PORT),
139:        DEVICE_NUMBER_IN_USE(0x08, "Device Number In Use", DeviceFinder.ANNOUNCEMENT_PORT),
145:        CDJ_STATUS(0x0a, "CDJ Status", VirtualCdj.UPDATE_PORT),
184:        PRECISE_POSITION(0x0b, "Precise Position", BeatFinder.BEAT_PORT),
Shell cwd was reset to /Users/noah/conductor/workspaces/deadca7/san-diego


 I'm mapping out the type bytes for the device negotiation protocol on port 50000, tracking the handshake sequence where the mixer assigns device numbers through a multi-stage process.

The key insight is that during the claim handshake, I need to listen for incoming packets on port 50000 to detect if the mixer assigns a number or if another device is already using the one I'm claiming. After going online, I also need to defend against other devices trying to claim my number.

The architectural challenge is that the capture tool and the VCDJ both need to read from the same UDP connection on port 50000, but UDP sockets can only deliver each datagram to one reader. I'm considering having the VCDJ own the receive loop on port 50000 and expose a callback so the capture tool can still record those packets, keeping the library reusable while maintaining the capture-first testing approach.

For the implementation, I'll separate concerns: pure helper functions for self-assignment logic and packet parsing (which are unit-testable), and a live `Run` method that orchestrates the full negotiation loop with concurrent reading and reacting. The pure helpers include determining the next unused device number, classifying incoming packets to detect defense or mixer assignments, and building defense and assignment-request packets. The `Run` method will spawn a reader goroutine that processes incoming datagrams, updates shared state atomically, and forwards packets to a callback for recording, while the main flow executes the claim handshake stages and watches for rejection or assignment.

For now I'll default to a fixed device number (5) which we've validated, with auto-assign (Device==0) as an option that watches the network for used numbers before claiming.

I'm restructuring the API: replacing the simple `Announce` and `KeepAlive` methods with a new `Online` method that takes the context, connection, and IP network, computes the broadcast address, and runs the full negotiation and keepalive loop. This consolidates the logic and makes the reader ownership clear. The low-level packet builders stay, but the claim logic becomes private and reactive. I'll update the capture tool to use `Online` instead of calling `Announce` separately, and adjust its listener so it doesn't duplicate reads on the 50000 port.

Defining the runtime state with atomics for rejection and mixer assignment, plus a concurrent map to track seen device numbers from keepalives. The handler will react to incoming packets: recording device numbers from keepalives, acknowledging mixer offers, storing assigned numbers, and sending defense packets when needed.

Defining the handler logic... I need to be careful about copying VCDJ by value since it contains atomic fields, which shouldn't be copied. The unexported fields themselves are fine with zero values, but I need to ensure the struct isn't passed around by value in tests and other places. Atomic types like Bool and Int32 don't have noCopy, so vet won't flag them, but semantically it's cleaner to keep runtime state out of VCDJ entirely. I'll create an internal state struct inside Online with the atomic fields and closure handlers that capture both the state and the VCDJ pointer. For the mixer assignment logic, I'll implement the full WILL_ASSIGN/ASSIGN reactions to handle auto-assign properly, defaulting to fixed device 5 but supporting Device 0 for dynamic assignment.

Now I'm generating the hex payloads for the assignRequest and defense messages.


I now have the full negotiation protocol. The mixer's flow on port 50000 uses distinct type bytes: `0x01` will-assign, `0x03` assign, `0x05` assignment-finished (that's the `0x05` we saw the DJM send), `0x08` defense (device-in-use). Let me get the two remaining packet templates.


Generate assign-request and defense template hex


assignRequest  len=50 (0x32)  "5173707431576d4a4f4c0201626561742d6c696e6b0000000000000000000000010200320000000000000000000000000100"
defense        len=41 (0x29)  "5173707431576d4a4f4c0800626561742d6c696e6b0000000000000000000000010200290000000000"


 I'm redesigning announce.go to include the new negotiation packet types and expanding the template system with assignment and defense packet templates. I'll add two new builder functions with value receivers for creating these packets, implement a SelfAssign method that picks an available byte from a usage map, and create a reactive Online method that manages the connection lifecycle including reading, claiming, keepalive, and defense operations.

For Online, I'm using a pointer receiver since it mutates the Device field during self-assignment. The method will own the UDP connection and handle the full negotiation flow: sending hello packets, determining which device number to claim (either the configured one or auto-assigning from available numbers), running the three-stage claim sequence with rejection handling, and then maintaining the online state with periodic keepalive packets while the reader goroutine continues defending against other devices trying to claim the same number.

I'm also refactoring the capture tool to use Online instead of the old Announce and KeepAlive methods, passing a callback that feeds packets into the channel, and updating the discovery function to return the IP network alongside the VCDJ instance so Online can properly initialize the broadcast address and IP field.


Now I'll rewrite `announce.go` to make the virtual CDJ reactive: a `Online` method that owns port 50000 — running the claim handshake while concurrently listening for **defenses** (`0x08`, retry/abort) and **mixer assignments** (`0x01`/`0x03`/`0x05`, accept the mixer's number), then defending its own number once online. Plus a pure, testable `SelfAssign`.


Write

package link

import (
	"bytes"
	"context"
	"encoding/hex"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
)

// Packet types exchanged while a virtual CDJ comes online, interpreted on port
// 50000. A player sends hello, then three device-number claim stages, before
// keep-alives; a mixer may interrupt to assign a number, and any device
// defends a number it already owns. See
// https://djl-analysis.deepsymmetry.org/djl-analysis/startup.html#cdj-startup
const (
	TypeClaim1         Type = 0x00
	TypeWillAssign     Type = 0x01
	TypeClaim2         Type = 0x02
	TypeAssign         Type = 0x03
	TypeClaim3         Type = 0x04
	TypeAssignFinished Type = 0x05
	TypeDefense        Type = 0x08
	TypeHello          Type = 0x0a
)

// ClaimInterval is the gap between packets within a startup stage.
const ClaimInterval = 300 * time.Millisecond

// KeepAliveInterval is the gap between keep-alive packets once online.
const KeepAliveInterval = 1500 * time.Millisecond

// WatchPeriod is how long to observe keep-alives before self-assigning, so we
// avoid claiming a number already in use.
const WatchPeriod = 2 * time.Second

// Packet templates copied verbatim from beat-link; per-device fields (name,
// number, MAC, IP, packet counter) are patched into a copy before sending.
var (
	tmplAssign    = mustHex("5173707431576d4a4f4c0201626561742d6c696e6b0000000000000000000000010200320000000000000000000000000100")
	tmplClaim1    = mustHex("5173707431576d4a4f4c0000626561742d6c696e6b00000000000000000000000103002c0d01000000000000")
	tmplClaim2    = mustHex("5173707431576d4a4f4c0200626561742d6c696e6b000000000000000000000001030032000000000000000000000d000100")
	tmplClaim3    = mustHex("5173707431576d4a4f4c0400626561742d6c696e6b0000000000000000000000010300260d00")
	tmplDefense   = mustHex("5173707431576d4a4f4c0800626561742d6c696e6b0000000000000000000000010200290000000000")
	tmplHello     = mustHex("5173707431576d4a4f4c0a00626561742d6c696e6b0000000000000000000000010400260140")
	tmplKeepAlive = mustHex("5173707431576d4a4f4c0600626561742d6c696e6b000000000000000000000001020036000100000000000000000000020000000164")
)

func mustHex(s string) []byte {
	b, err := hex.DecodeString(s)
	if err != nil {
		panic(err)
	}
	return b
}

// SelfAssign returns the lowest player number not in used that is greater than
// after, starting no lower than 5 (the first number a CDJ-3000 may take) and
// stopping below the 33+ range mixers use.
func SelfAssign(used map[byte]bool, after byte) (byte, bool) {
	n := after + 1
	if n < 5 {
		n = 5
	}
	for ; n <= 15; n++ {
		if !used[n] {
			return n, true
		}
	}
	return 0, false
}

// VCDJ is a virtual CDJ that participates in the network so real players send
// it status, metadata, and database access. Device 0 means self-assign.
type VCDJ struct {
	Device byte
	IP     net.IP
	MAC    net.HardwareAddr
	Name   string
}

// Online brings the virtual CDJ online and holds its device number until ctx is
// done. It owns conn (bound to port 50000): it runs the claim handshake while
// reacting to defenses and mixer assignments, then sends keep-alives and
// defends its number. Every received datagram is passed to onPacket so callers
// can record it.
func (v *VCDJ) Online(ctx context.Context, conn *net.UDPConn, ipNet *net.IPNet, onPacket func([]byte, *net.UDPAddr)) error {
	v.IP = ipNet.IP
	dst := &net.UDPAddr{IP: Broadcast(ipNet), Port: 50000}

	var (
		assigned atomic.Int32
		claiming atomic.Int32
		online   atomic.Bool
		rejected atomic.Bool
		seen     sync.Map
	)

	send := func(b []byte, to *net.UDPAddr) {
		conn.WriteToUDP(b, to)
	}

	handle := func(b []byte, src *net.UDPAddr) {
		if len(b) < 0x25 || !bytes.Equal(b[:len(Magic)], Magic) {
			return
		}
		peer := &net.UDPAddr{IP: src.IP, Port: 50000}
		switch Type(b[0x0a]) {
		case TypeKeepAlive:
			seen.Store(b[0x24], true)
		case TypeWillAssign:
			if claiming.Load() != 0 {
				send(v.assignRequest(), peer)
			}
		case TypeAssign:
			assigned.Store(int32(b[0x24]))
		case TypeAssignFinished:
			assigned.Store(claiming.Load())
		case TypeDefense:
			if b[0x24] == byte(claiming.Load()) {
				rejected.Store(true)
			}
		case TypeClaim2:
			if online.Load() && len(b) > 0x2e && b[0x2e] == v.Device {
				send(v.defense(), peer)
			}
		case TypeClaim3:
			if online.Load() && b[0x24] == v.Device {
				send(v.defense(), peer)
			}
		}
	}

	errc := make(chan error, 1)
	go func() {
		go func() { <-ctx.Done(); conn.SetReadDeadline(time.Now()) }()
		buf := make([]byte, 1500)
		for {
			n, src, err := conn.ReadFromUDP(buf)
			if err != nil {
				if ctx.Err() != nil {
					errc <- nil
				} else {
					errc <- errors.WithStack(err)
				}
				return
			}
			b := append([]byte(nil), buf[:n]...)
			handle(b, src)
			if onPacket != nil {
				onPacket(b, src)
			}
		}
	}()

	wait := func(d time.Duration) error {
		select {
		case <-ctx.Done():
			return errors.WithStack(ctx.Err())
		case err := <-errc:
			if err == nil {
				err = errors.New("listener stopped")
			}
			return err
		case <-time.After(d):
			return nil
		}
	}

	for i := 0; i < 3; i++ {
		send(v.hello(), dst)
		if err := wait(ClaimInterval); err != nil {
			return err
		}
	}

	auto := v.Device == 0
	if auto {
		if err := wait(WatchPeriod); err != nil {
			return err
		}
	}

	claim := func(num byte) (bool, error) {
		v.Device = num
		claiming.Store(int32(num))
		rejected.Store(false)
		assigned.Store(0)

		stages := [][]func(byte) []byte{
			{v.claim1}, {v.claim2}, {v.claim3},
		}
		for s, stage := range stages {
			if s == 2 {
				if n := assigned.Swap(0); n > 0 {
					v.Device = byte(n)
					claiming.Store(n)
				}
			}
			for i := byte(1); i <= 3 && assigned.Load() == 0; i++ {
				send(stage[0](i), dst)
				if err := wait(ClaimInterval); err != nil {
					return false, err
				}
				if rejected.Load() {
					return false, nil
				}
			}
		}
		return true, nil
	}

	var after byte
	for {
		num := v.Device
		if auto {
			n, ok := SelfAssign(usedNumbers(&seen), after)
			if !ok {
				return errors.New("no available device number")
			}
			num, after = n, n
		}

		ok, err := claim(num)
		if err != nil {
			return err
		}
		if ok {
			break
		}
		if !auto {
			return errors.Errorf("device number %d defended by another device", num)
		}
	}

	online.Store(true)
	claiming.Store(0)

	t := time.NewTicker(KeepAliveInterval)
	defer t.Stop()
	for {
		send(v.keepAlive(), dst)
		select {
		case <-ctx.Done():
			return nil
		case err := <-errc:
			return err
		case <-t.C:
		}
	}
}

func usedNumbers(seen *sync.Map) map[byte]bool {
	used := map[byte]bool{}
	seen.Range(func(k, _ any) bool {
		used[k.(byte)] = true
		return true
	})
	return used
}

func (v VCDJ) name(b []byte) {
	for i := 0; i < 0x14; i++ {
		b[0x0c+i] = 0
	}
	copy(b[0x0c:0x0c+0x14], v.Name)
}

func (v VCDJ) hello() []byte {
	b := append([]byte(nil), tmplHello...)
	v.name(b)
	return b
}

func (v VCDJ) claim1(counter byte) []byte {
	b := append([]byte(nil), tmplClaim1...)
	v.name(b)
	b[0x24] = counter
	copy(b[0x26:0x2c], v.MAC)
	return b
}

func (v VCDJ) claim2(counter byte) []byte {
	b := append([]byte(nil), tmplClaim2...)
	v.name(b)
	copy(b[0x24:0x28], v.IP.To4())
	copy(b[0x28:0x2e], v.MAC)
	b[0x2e] = v.Device
	b[0x2f] = counter
	b[0x31] = v.assignFlag()
	return b
}

func (v VCDJ) claim3(counter byte) []byte {
	b := append([]byte(nil), tmplClaim3...)
	v.name(b)
	b[0x24] = v.Device
	b[0x25] = counter
	return b
}

func (v VCDJ) assignRequest() []byte {
	b := append([]byte(nil), tmplAssign...)
	v.name(b)
	copy(b[0x24:0x28], v.IP.To4())
	copy(b[0x28:0x2e], v.MAC)
	b[0x2f] = 0x01
	b[0x31] = v.assignFlag()
	return b
}

func (v VCDJ) defense() []byte {
	b := append([]byte(nil), tmplDefense...)
	v.name(b)
	b[0x24] = v.Device
	copy(b[0x25:0x29], v.IP.To4())
	return b
}

func (v VCDJ) keepAlive() []byte {
	b := append([]byte(nil), tmplKeepAlive...)
	v.name(b)
	b[0x24] = v.Device
	copy(b[0x26:0x2c], v.MAC)
	copy(b[0x2c:0x30], v.IP.To4())
	return b
}

func (v VCDJ) assignFlag() byte {
	if v.Device == 0 {
		return 0x01
	}
	return 0x02
}


The file /Users/noah/conductor/workspaces/deadca7/san-diego/link/announce.go has been updated successfully. (file state is current in your context — no need to Read it back)

package link

import (
	"bytes"
	"context"
	"encoding/hex"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
)

// Packet types exchanged while a virtual CDJ comes online, interpreted on port
// 50000. A player sends hello, then three device-number claim stages, before
// keep-alives; a mixer may interrupt to assign a number, and any device
// defends a number it already owns. See
// https://djl-analysis.deepsymmetry.org/djl-analysis/startup.html#cdj-startup
const (
	TypeClaim1         Type = 0x00
	TypeWillAssign     Type = 0x01
	TypeClaim2         Type = 0x02
	TypeAssign         Type = 0x03
	TypeClaim3         Type = 0x04
	TypeAssignFinished Type = 0x05
	TypeDefense        Type = 0x08
	TypeHello          Type = 0x0a
)

// ClaimInterval is the gap between packets within a startup stage.
const ClaimInterval = 300 * time.Millisecond

// KeepAliveInterval is the gap between keep-alive packets once online.
const KeepAliveInterval = 1500 * time.Millisecond

// WatchPeriod is how long to observe keep-alives before self-assigning, so we
// avoid claiming a number already in use.
const WatchPeriod = 2 * time.Second

// Packet templates copied verbatim from beat-link; per-device fields (name,
// number, MAC, IP, packet counter) are patched into a copy before sending.
var (
	tmplAssign    = mustHex("5173707431576d4a4f4c0201626561742d6c696e6b0000000000000000000000010200320000000000000000000000000100")
	tmplClaim1    = mustHex("5173707431576d4a4f4c0000626561742d6c696e6b00000000000000000000000103002c0d01000000000000")
	tmplClaim2    = mustHex("5173707431576d4a4f4c0200626561742d6c696e6b000000000000000000000001030032000000000000000000000d000100")
	tmplClaim3    = mustHex("5173707431576d4a4f4c0400626561742d6c696e6b0000000000000000000000010300260d00")
	tmplDefense   = mustHex("5173707431576d4a4f4c0800626561742d6c696e6b0000000000000000000000010200290000000000")
	tmplHello     = mustHex("5173707431576d4a4f4c0a00626561742d6c696e6b0000000000000000000000010400260140")
	tmplKeepAlive = mustHex("5173707431576d4a4f4c0600626561742d6c696e6b000000000000000000000001020036000100000000000000000000020000000164")
)

func mustHex(s string) []byte {
	b, err := hex.DecodeString(s)
	if err != nil {
		panic(err)
	}
	return b
}

// SelfAssign returns the lowest player number not in used that is greater than
// after, starting no lower than 5 (the first number a CDJ-3000 may take) and
// stopping below the 33+ range mixers use.
func SelfAssign(used map[byte]bool, after byte) (byte, bool) {
	n := after + 1
	if n < 5 {
		n = 5
	}
	for ; n <= 15; n++ {
		if !used[n] {
			return n, true
		}
	}
	return 0, false
}

// VCDJ is a virtual CDJ that participates in the network so real players send
// it status, metadata, and database access. Device 0 means self-assign.
type VCDJ struct {
	Device byte
	IP     net.IP
	MAC    net.HardwareAddr
	Name   string
}

// Online brings the virtual CDJ online and holds its device number until ctx is
// done. It owns conn (bound to port 50000): it runs the claim handshake while
// reacting to defenses and mixer assignments, then sends keep-alives and
// defends its number. Every received datagram is passed to onPacket so callers
// can record it.
func (v *VCDJ) Online(ctx context.Context, conn *net.UDPConn, ipNet *net.IPNet, onPacket func([]byte, *net.UDPAddr)) error {
	v.IP = ipNet.IP
	dst := &net.UDPAddr{IP: Broadcast(ipNet), Port: 50000}

	var (
		assigned atomic.Int32
		claiming atomic.Int32
		online   atomic.Bool
		rejected atomic.Bool
		seen     sync.Map
	)

	send := func(b []byte, to *net.UDPAddr) {
		conn.WriteToUDP(b, to)
	}

	handle := func(b []byte, src *net.UDPAddr) {
		if len(b) < 0x25 || !bytes.Equal(b[:len(Magic)], Magic) {
			return
		}
		peer := &net.UDPAddr{IP: src.IP, Port: 50000}
		switch Type(b[0x0a]) {
		case TypeKeepAlive:
			seen.Store(b[0x24], true)
		case TypeWillAssign:
			if claiming.Load() != 0 {
				send(v.assignRequest(), peer)
			}
		case TypeAssign:
			assigned.Store(int32(b[0x24]))
		case TypeAssignFinished:
			assigned.Store(claiming.Load())
		case TypeDefense:
			if b[0x24] == byte(claiming.Load()) {
				rejected.Store(true)
			}
		case TypeClaim2:
			if online.Load() && len(b) > 0x2e && b[0x2e] == v.Device {
				send(v.defense(), peer)
			}
		case TypeClaim3:
			if online.Load() && b[0x24] == v.Device {
				send(v.defense(), peer)
			}
		}
	}

	errc := make(chan error, 1)
	go func() {
		go func() { <-ctx.Done(); conn.SetReadDeadline(time.Now()) }()
		buf := make([]byte, 1500)
		for {
			n, src, err := conn.ReadFromUDP(buf)
			if err != nil {
				if ctx.Err() != nil {
					errc <- nil
				} else {
					errc <- errors.WithStack(err)
				}
				return
			}
			b := append([]byte(nil), buf[:n]...)
			handle(b, src)
			if onPacket != nil {
				onPacket(b, src)
			}
		}
	}()

	wait := func(d time.Duration) error {
		select {
		case <-ctx.Done():
			return errors.WithStack(ctx.Err())
		case err := <-errc:
			if err == nil {
				err = errors.New("listener stopped")
			}
			return err
		case <-time.After(d):
			return nil
		}
	}

	for i := 0; i < 3; i++ {
		send(v.hello(), dst)
		if err := wait(ClaimInterval); err != nil {
			return err
		}
	}

	auto := v.Device == 0
	if auto {
		if err := wait(WatchPeriod); err != nil {
			return err
		}
	}

	claim := func(num byte) (bool, error) {
		v.Device = num
		claiming.Store(int32(num))
		rejected.Store(false)
		assigned.Store(0)

		stages := [][]func(byte) []byte{
			{v.claim1}, {v.claim2}, {v.claim3},
		}
		for s, stage := range stages {
			if s == 2 {
				if n := assigned.Swap(0); n > 0 {
					v.Device = byte(n)
					claiming.Store(n)
				}
			}
			for i := byte(1); i <= 3 && assigned.Load() == 0; i++ {
				send(stage[0](i), dst)
				if err := wait(ClaimInterval); err != nil {
					return false, err
				}
				if rejected.Load() {
					return false, nil
				}
			}
		}
		return true, nil
	}

	var after byte
	for {
		num := v.Device
		if auto {
			n, ok := SelfAssign(usedNumbers(&seen), after)
			if !ok {
				return errors.New("no available device number")
			}
			num, after = n, n
		}

		ok, err := claim(num)
		if err != nil {
			return err
		}
		if ok {
			break
		}
		if !auto {
			return errors.Errorf("device number %d defended by another device", num)
		}
	}

	online.Store(true)
	claiming.Store(0)

	t := time.NewTicker(KeepAliveInterval)
	defer t.Stop()
	for {
		send(v.keepAlive(), dst)
		select {
		case <-ctx.Done():
			return nil
		case err := <-errc:
			return err
		case <-t.C:
		}
	}
}

func usedNumbers(seen *sync.Map) map[byte]bool {
	used := map[byte]bool{}
	seen.Range(func(k, _ any) bool {
		used[k.(byte)] = true
		return true
	})
	return used
}

func (v VCDJ) name(b []byte) {
	for i := 0; i < 0x14; i++ {
		b[0x0c+i] = 0
	}
	copy(b[0x0c:0x0c+0x14], v.Name)
}

func (v VCDJ) hello() []byte {
	b := append([]byte(nil), tmplHello...)
	v.name(b)
	return b
}

func (v VCDJ) claim1(counter byte) []byte {
	b := append([]byte(nil), tmplClaim1...)
	v.name(b)
	b[0x24] = counter
	copy(b[0x26:0x2c], v.MAC)
	return b
}

func (v VCDJ) claim2(counter byte) []byte {
	b := append([]byte(nil), tmplClaim2...)
	v.name(b)
	copy(b[0x24:0x28], v.IP.To4())
	copy(b[0x28:0x2e], v.MAC)
	b[0x2e] = v.Device
	b[0x2f] = counter
	b[0x31] = v.assignFlag()
	return b
}

func (v VCDJ) claim3(counter byte) []byte {
	b := append([]byte(nil), tmplClaim3...)
	v.name(b)
	b[0x24] = v.Device
	b[0x25] = counter
	return b
}

func (v VCDJ) assignRequest() []byte {
	b := append([]byte(nil), tmplAssign...)
	v.name(b)
	copy(b[0x24:0x28], v.IP.To4())
	copy(b[0x28:0x2e], v.MAC)
	b[0x2f] = 0x01
	b[0x31] = v.assignFlag()
	return b
}

func (v VCDJ) defense() []byte {
	b := append([]byte(nil), tmplDefense...)
	v.name(b)
	b[0x24] = v.Device
	copy(b[0x25:0x29], v.IP.To4())
	return b
}

func (v VCDJ) keepAlive() []byte {
	b := append([]byte(nil), tmplKeepAlive...)
	v.name(b)
	b[0x24] = v.Device
	copy(b[0x26:0x2c], v.MAC)
	copy(b[0x2c:0x30], v.IP.To4())
	return b
}

func (v VCDJ) assignFlag() byte {
	if v.Device == 0 {
		return 0x01
	}
	return 0x02
}

-// Packet types sent while a virtual CDJ comes online, in order. A real player

-// sends hello, then three device-number claim stages, before keep-alives; see

+// Packet types exchanged while a virtual CDJ comes online, interpreted on port

+// 50000. A player sends hello, then three device-number claim stages, before

+// keep-alives; a mixer may interrupt to assign a number, and any device

+// defends a number it already owns. See

+// WatchPeriod is how long to observe keep-alives before self-assigning, so we

+// avoid claiming a number already in use.

+// SelfAssign returns the lowest player number not in used that is greater than

+// after, starting no lower than 5 (the first number a CDJ-3000 may take) and

+// stopping below the 33+ range mixers use.

+func SelfAssign(used map[byte]bool, after byte) (byte, bool) {

-// it status, metadata, and database access.

+// it status, metadata, and database access. Device 0 means self-assign.

-// Announce performs the four-stage startup handshake on conn (bound to port

-// 50000), broadcasting each packet to dst. It blocks for the full sequence

-// (~3.6s) so a caller can treat its return as "online".

-func (v VCDJ) Announce(ctx context.Context, conn *net.UDPConn, dst *net.UDPAddr) error {

-  pkts = append(pkts, v.hello(), v.hello(), v.hello())

+// Online brings the virtual CDJ online and holds its device number until ctx is

+// done. It owns conn (bound to port 50000): it runs the claim handshake while

+// reacting to defenses and mixer assignments, then sends keep-alives and

+// defends its number. Every received datagram is passed to onPacket so callers

+func (v *VCDJ) Online(ctx context.Context, conn *net.UDPConn, ipNet *net.IPNet, onPacket func([]byte, *net.UDPAddr)) error {

+  dst := &net.UDPAddr{IP: Broadcast(ipNet), Port: 50000}

+  send := func(b []byte, to *net.UDPAddr) {

+  handle := func(b []byte, src *net.UDPAddr) {

+    if len(b) < 0x25 || !bytes.Equal(b[:len(Magic)], Magic) {

+    peer := &net.UDPAddr{IP: src.IP, Port: 50000}

+      if b[0x24] == byte(claiming.Load()) {

+      if online.Load() && len(b) > 0x2e && b[0x2e] == v.Device {

+      if online.Load() && b[0x24] == v.Device {

-    if _, err := conn.WriteToUDP(p, dst); err != nil {

+    go func() { <-ctx.Done(); conn.SetReadDeadline(time.Now()) }()

+      n, src, err := conn.ReadFromUDP(buf)

+          errc <- errors.WithStack(err)

+      b := append([]byte(nil), buf[:n]...)

+  wait := func(d time.Duration) error {

+        err = errors.New("listener stopped")

-// KeepAlive broadcasts a keep-alive immediately and then every

-// KeepAliveInterval until ctx is done, holding the claimed device number.

-func (v VCDJ) KeepAlive(ctx context.Context, conn *net.UDPConn, dst *net.UDPAddr) error {

+    if err := wait(ClaimInterval); err != nil {

+    if err := wait(WatchPeriod); err != nil {

+  claim := func(num byte) (bool, error) {

+      {v.claim1}, {v.claim2}, {v.claim3},

+        if n := assigned.Swap(0); n > 0 {

+      for i := byte(1); i <= 3 && assigned.Load() == 0; i++ {

+      n, ok := SelfAssign(usedNumbers(&seen), after)

+        return errors.New("no available device number")

+      return errors.Errorf("device number %d defended by another device", num)

-    if _, err := conn.WriteToUDP(v.keepAlive(), dst); err != nil {

+func usedNumbers(seen *sync.Map) map[byte]bool {

copy(b[0x28:0x2e], v.MAC)
b[0x2e] = v.Device
b[0x2f] = counter
-  b[0x31] = 0x02
+  b[0x31] = v.assignFlag()
return b
}

+  b := append([]byte(nil), tmplAssign...)

+  b := append([]byte(nil), tmplDefense...)

copy(b[0x2c:0x30], v.IP.To4())
return b
}
+
+func (v VCDJ) assignFlag() byte {
+  if v.Device == 0 {
+    return 0x01
+  }
+  return 0x02
+}

package link

import (
	"context"
	"encoding/hex"
	"net"
	"time"

	"github.com/cockroachdb/errors"
)

// Packet types sent while a virtual CDJ comes online, in order. A real player
// sends hello, then three device-number claim stages, before keep-alives; see
// https://djl-analysis.deepsymmetry.org/djl-analysis/startup.html#cdj-startup
const (
	TypeClaim1  Type = 0x00
	TypeClaim2  Type = 0x02
	TypeClaim3  Type = 0x04
	TypeDefense Type = 0x08
	TypeHello   Type = 0x0a
)

// ClaimInterval is the gap between packets within a startup stage.
const ClaimInterval = 300 * time.Millisecond

// KeepAliveInterval is the gap between keep-alive packets once online.
const KeepAliveInterval = 1500 * time.Millisecond

// Packet templates copied verbatim from beat-link; per-device fields (name,
// number, MAC, IP, packet counter) are patched into a copy before sending.
var (
	tmplClaim1    = mustHex("5173707431576d4a4f4c0000626561742d6c696e6b00000000000000000000000103002c0d01000000000000")
	tmplClaim2    = mustHex("5173707431576d4a4f4c0200626561742d6c696e6b000000000000000000000001030032000000000000000000000d000100")
	tmplClaim3    = mustHex("5173707431576d4a4f4c0400626561742d6c696e6b0000000000000000000000010300260d00")
	tmplHello     = mustHex("5173707431576d4a4f4c0a00626561742d6c696e6b0000000000000000000000010400260140")
	tmplKeepAlive = mustHex("5173707431576d4a4f4c0600626561742d6c696e6b000000000000000000000001020036000100000000000000000000020000000164")
)

func mustHex(s string) []byte {
	b, err := hex.DecodeString(s)
	if err != nil {
		panic(err)
	}
	return b
}

// VCDJ is a virtual CDJ that participates in the network so real players send
// it status, metadata, and database access.
type VCDJ struct {
	Device byte
	IP     net.IP
	MAC    net.HardwareAddr
	Name   string
}

// Announce performs the four-stage startup handshake on conn (bound to port
// 50000), broadcasting each packet to dst. It blocks for the full sequence
// (~3.6s) so a caller can treat its return as "online".
func (v VCDJ) Announce(ctx context.Context, conn *net.UDPConn, dst *net.UDPAddr) error {
	var pkts [][]byte
	pkts = append(pkts, v.hello(), v.hello(), v.hello())
	for i := byte(1); i <= 3; i++ {
		pkts = append(pkts, v.claim1(i))
	}
	for i := byte(1); i <= 3; i++ {
		pkts = append(pkts, v.claim2(i))
	}
	for i := byte(1); i <= 3; i++ {
		pkts = append(pkts, v.claim3(i))
	}

	for _, p := range pkts {
		if _, err := conn.WriteToUDP(p, dst); err != nil {
			return errors.WithStack(err)
		}
		select {
		case <-ctx.Done():
			return errors.WithStack(ctx.Err())
		case <-time.After(ClaimInterval):
		}
	}
	return nil
}

// KeepAlive broadcasts a keep-alive immediately and then every
// KeepAliveInterval until ctx is done, holding the claimed device number.
func (v VCDJ) KeepAlive(ctx context.Context, conn *net.UDPConn, dst *net.UDPAddr) error {
	t := time.NewTicker(KeepAliveInterval)
	defer t.Stop()
	for {
		if _, err := conn.WriteToUDP(v.keepAlive(), dst); err != nil {
			return errors.WithStack(err)
		}
		select {
		case <-ctx.Done():
			return nil
		case <-t.C:
		}
	}
}

func (v VCDJ) name(b []byte) {
	for i := 0; i < 0x14; i++ {
		b[0x0c+i] = 0
	}
	copy(b[0x0c:0x0c+0x14], v.Name)
}

func (v VCDJ) hello() []byte {
	b := append([]byte(nil), tmplHello...)
	v.name(b)
	return b
}

func (v VCDJ) claim1(counter byte) []byte {
	b := append([]byte(nil), tmplClaim1...)
	v.name(b)
	b[0x24] = counter
	copy(b[0x26:0x2c], v.MAC)
	return b
}

func (v VCDJ) claim2(counter byte) []byte {
	b := append([]byte(nil), tmplClaim2...)
	v.name(b)
	copy(b[0x24:0x28], v.IP.To4())
	copy(b[0x28:0x2e], v.MAC)
	b[0x2e] = v.Device
	b[0x2f] = counter
	b[0x31] = 0x02
	return b
}

func (v VCDJ) claim3(counter byte) []byte {
	b := append([]byte(nil), tmplClaim3...)
	v.name(b)
	b[0x24] = v.Device
	b[0x25] = counter
	return b
}

func (v VCDJ) keepAlive() []byte {
	b := append([]byte(nil), tmplKeepAlive...)
	v.name(b)
	b[0x24] = v.Device
	copy(b[0x26:0x2c], v.MAC)
	copy(b[0x2c:0x30], v.IP.To4())
	return b
}


I need to refactor the capture tool to use the Online discovery method instead of Announce+KeepAlive, and skip the listen call when announcing. I'm also updating the online() helper to return the IP network instead of the destination address.


Now update the capture tool to use the new reactive `Online` (and stop double-reading port 50000 when announcing):


Edit

	var vcdj *link.VCDJ
	var dst *net.UDPAddr
	if announce {
		vcdj, dst, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			if err := vcdj.Announce(ctx, c0, dst); err != nil {
				return err
			}
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return vcdj.KeepAlive(ctx, c0, dst)
		})
	}

	var vcdj *link.VCDJ
	var ipNet *net.IPNet
	if announce {
		vcdj, ipNet, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	record := func(b []byte, src *net.UDPAddr) {
		select {
		case pkts <- packet(b, 50000, src):
		case <-ctx.Done():
		}
	}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			err := vcdj.Online(ctx, c0, ipNet, record)
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return err
		})
	} else {
		g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	}


	var vcdj *link.VCDJ
	var ipNet *net.IPNet
	if announce {
		vcdj, ipNet, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	record := func(b []byte, src *net.UDPAddr) {
		select {
		case pkts <- packet(b, 50000, src):
		case <-ctx.Done():
		}
	}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			err := vcdj.Online(ctx, c0, ipNet, record)
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return err
		})
	} else {
		g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	}

// Command capture records raw Pro DJ Link UDP datagrams to a JSONL fixture.
//
// It listens on the DJ Link discovery, beat, and status ports and writes one
// Packet per line so tests can replay real traffic without hardware. With
// -announce it first brings a virtual CDJ online (the full startup handshake)
// so real players send status, metadata, and database traffic to it.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

// Packet is one recorded UDP datagram.
type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	announce := flag.Bool("announce", false, "bring a virtual CDJ online while capturing")
	device := flag.Int("device", 5, "virtual CDJ device number")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "capture.jsonl", "output JSONL fixture path")
	secs := flag.Int("secs", 8, "capture duration in seconds")
	flag.Parse()

	if err := run(*out, time.Duration(*secs)*time.Second, *announce, byte(*device), *name); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

func run(out string, d time.Duration, announce bool, device byte, name string) error {
	f, err := os.Create(out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, d)
	defer cancel()

	c0, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50000})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}

	c1, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50001})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c1.Close()

	c2, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50002})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c2.Close()

	var vcdj *link.VCDJ
	var dst *net.UDPAddr
	if announce {
		vcdj, dst, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			if err := vcdj.Announce(ctx, c0, dst); err != nil {
				return err
			}
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return vcdj.KeepAlive(ctx, c0, dst)
		})
	}

	g.Go(func() error {
		n := 0
		for {
			select {
			case <-ctx.Done():
				fmt.Fprintf(os.Stderr, "captured %d packets to %s\n", n, out)
				return nil
			case p := <-pkts:
				if err := enc.Encode(p); err != nil {
					return errors.WithStack(err)
				}
				n++
			}
		}
	})

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// online discovers a real device to learn the local interface, then returns a
// virtual CDJ and the broadcast destination for its announce packets.
func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.UDPAddr, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})

	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}

		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{
			Device: device,
			IP:     ipNet.IP,
			MAC:    iface.HardwareAddr,
			Name:   name,
		}, &net.UDPAddr{IP: link.Broadcast(ipNet), Port: 50000}, nil
	}
}

func listen(ctx context.Context, c *net.UDPConn, port int, pkts chan<- Packet) error {
	go func() {
		<-ctx.Done()
		c.SetReadDeadline(time.Now())
	}()

	buf := make([]byte, 1500)
	for {
		n, addr, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		select {
		case pkts <- Packet{
			Data: fmt.Sprintf("%x", buf[:n]),
			Len:  n,
			Port: port,
			Src:  addr.String(),
			TS:   time.Now().UnixMilli(),
		}:
		case <-ctx.Done():
			return nil
		}
	}
}

-    vcdj, dst, err = online(c0, device, name)

+    vcdj, ipNet, err = online(c0, device, name)

+  record := func(b []byte, src *net.UDPAddr) {

+    case pkts <- packet(b, 50000, src):

-  g.Go(func() error { return listen(ctx, c0, 50000, pkts) })

-      if err := vcdj.Announce(ctx, c0, dst); err != nil {

+      err := vcdj.Online(ctx, c0, ipNet, record)

-      return vcdj.KeepAlive(ctx, c0, dst)

+    g.Go(func() error { return listen(ctx, c0, 50000, pkts) })


 The logging happens after Online returns, which is misleading since the device number gets assigned inside Online. I should either drop the log or keep it at the end to show the final device number that was claimed.

Now I need to refactor the packet helper and update the online return type. I'm adding a packet constructor function that takes a byte slice, port, and source address, then using it consistently in both listen and record. I also need to update online to return the IP network information.


Let me add the `packet` helper, fix the log, and update `online()` to return the `ipNet`:


		g.Go(func() error {
			err := vcdj.Online(ctx, c0, ipNet, record)
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return err
		})

		g.Go(func() error {
			err := vcdj.Online(ctx, c0, ipNet, record)
			fmt.Fprintf(os.Stderr, "virtual CDJ %q ran as device %d\n", vcdj.Name, vcdj.Device)
			return err
		})


		g.Go(func() error {
			err := vcdj.Online(ctx, c0, ipNet, record)
			fmt.Fprintf(os.Stderr, "virtual CDJ %q ran as device %d\n", vcdj.Name, vcdj.Device)
			return err
		})

// Command capture records raw Pro DJ Link UDP datagrams to a JSONL fixture.
//
// It listens on the DJ Link discovery, beat, and status ports and writes one
// Packet per line so tests can replay real traffic without hardware. With
// -announce it first brings a virtual CDJ online (the full startup handshake)
// so real players send status, metadata, and database traffic to it.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

// Packet is one recorded UDP datagram.
type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	announce := flag.Bool("announce", false, "bring a virtual CDJ online while capturing")
	device := flag.Int("device", 5, "virtual CDJ device number")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "capture.jsonl", "output JSONL fixture path")
	secs := flag.Int("secs", 8, "capture duration in seconds")
	flag.Parse()

	if err := run(*out, time.Duration(*secs)*time.Second, *announce, byte(*device), *name); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

func run(out string, d time.Duration, announce bool, device byte, name string) error {
	f, err := os.Create(out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, d)
	defer cancel()

	c0, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50000})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}

	c1, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50001})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c1.Close()

	c2, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 50002})
	if err != nil {
		return errors.WithStack(err)
	}
	defer c2.Close()

	var vcdj *link.VCDJ
	var ipNet *net.IPNet
	if announce {
		vcdj, ipNet, err = online(c0, device, name)
		if err != nil {
			return err
		}
	}

	enc := json.NewEncoder(f)
	pkts := make(chan Packet)
	record := func(b []byte, src *net.UDPAddr) {
		select {
		case pkts <- packet(b, 50000, src):
		case <-ctx.Done():
		}
	}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return listen(ctx, c1, 50001, pkts) })
	g.Go(func() error { return listen(ctx, c2, 50002, pkts) })

	if announce {
		g.Go(func() error {
			err := vcdj.Online(ctx, c0, ipNet, record)
			fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)
			return err
		})
	} else {
		g.Go(func() error { return listen(ctx, c0, 50000, pkts) })
	}

	g.Go(func() error {
		n := 0
		for {
			select {
			case <-ctx.Done():
				fmt.Fprintf(os.Stderr, "captured %d packets to %s\n", n, out)
				return nil
			case p := <-pkts:
				if err := enc.Encode(p); err != nil {
					return errors.WithStack(err)
				}
				n++
			}
		}
	})

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// online discovers a real device to learn the local interface, then returns a
// virtual CDJ and the broadcast destination for its announce packets.
func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.UDPAddr, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})

	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}

		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{
			Device: device,
			IP:     ipNet.IP,
			MAC:    iface.HardwareAddr,
			Name:   name,
		}, &net.UDPAddr{IP: link.Broadcast(ipNet), Port: 50000}, nil
	}
}

func listen(ctx context.Context, c *net.UDPConn, port int, pkts chan<- Packet) error {
	go func() {
		<-ctx.Done()
		c.SetReadDeadline(time.Now())
	}()

	buf := make([]byte, 1500)
	for {
		n, addr, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		select {
		case pkts <- Packet{
			Data: fmt.Sprintf("%x", buf[:n]),
			Len:  n,
			Port: port,
			Src:  addr.String(),
			TS:   time.Now().UnixMilli(),
		}:
		case <-ctx.Done():
			return nil
		}
	}
}

-      fmt.Fprintf(os.Stderr, "online as %q device %d\n", vcdj.Name, vcdj.Device)

+      fmt.Fprintf(os.Stderr, "virtual CDJ %q ran as device %d\n", vcdj.Name, vcdj.Device)