explewd/flit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix file transfer, signaling reliability, and DC callback races

Browse Source 
- wireFileRecv: replace OnMessage/OnClose channel approach with pion
  Detach API so SCTP buffers data regardless of callback timing — fixes
  0-byte files caused by pion dispatching data goroutines before OnMessage
  was registered
- yaw DC: same Detach API treatment for the control channel
- connectTo: use DetachDataChannels SettingEngine, replace stale PCs
  (Connecting/Failed/Disconnected state) instead of returning early;
  isCurrent guard in OnConnectionStateChange prevents stale-PC close
  from firing OnDisconnected
- offerByOrder: gate maybeOffer on peer ID order so only one side
  creates the yaw DC and SDP offer — fixes signaling glare causing
  triple connected (verified) logs
- pendingRecvs: promote single pendingRecv slot to map[string]*recvState
  so concurrent file offers don't overwrite each other
- peer-left: stop calling resetPeer() so active file DCs aren't closed
  before data arrives
- signaling ping: wrap conn.Ping with 10s timeout context so a dead TCP
  connection is detected within 30s rather than hanging forever
- signaling read: 3-minute read deadline forces reconnect if anchor
  silently evicts the peer from the room (idle timeout)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Fredrik Johansson
2026-07-02 17:38:48 +02:00
parent 488431eaec
commit 0fe01e146c
6 changed files with 307 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

321
cli/internal/transport/transport.go
View File

@@ -15,6 +15,7 @@ import (
"io"
"log"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
@@ -105,6 +106,26 @@ func dialSignaling(ctx context.Context, url string, id *crypto.Identity, netHash
}
}
}()
// Keepalive: ping every 20s so proxies/NAT don't silently drop the connection.
go func() {
t := time.NewTicker(20 * time.Second)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
err := conn.Ping(pingCtx)
cancel()
if err != nil {
// Close so the read loop errors and OnDisconnected fires.
_ = conn.Close(websocket.StatusGoingAway, "ping failed")
return
}
}
}
}()
for {
var msg anchorMsg
@@ -145,18 +166,20 @@ type Session struct {
identity *crypto.Identity
cfg Config
mu sync.Mutex
pc *webrtc.PeerConnection
dc *webrtc.DataChannel
peerID string
verified bool
esk, epk *[32]byte
peerEPK *[32]byte
ekeySent bool
offered bool
mu sync.Mutex
pc *webrtc.PeerConnection
dc *webrtc.DataChannel
peerID string
verified bool
esk, epk *[32]byte
peerEPK *[32]byte
ekeySent bool
offered bool
offerByOrder bool // true if this side should create the yaw DC and SDP offer
sig *signaling
pendingRecv *recvState
sig *signaling
pendingRecvs map[string]*recvState
pendingAccepts map[string]chan int64
OnConnected func(verified bool)
OnDisconnected func()
@@ -197,8 +220,18 @@ func (s *Session) Join(ctx context.Context, roomName string, trustedPeerID strin
go func() {
for {
// 3-minute read deadline: if the anchor silently evicts us from
// the room (idle timeout) without closing the WS, we'll never
// receive a peer-join. Force a reconnect so we re-enter the room.
readCtx, cancel := context.WithTimeout(ctx, 3*time.Minute)
var msg anchorMsg
if err := wsjson.Read(ctx, sig.conn, &msg); err != nil {
err := wsjson.Read(readCtx, sig.conn, &msg)
cancel()
if err != nil {
_ = sig.conn.Close(websocket.StatusGoingAway, "idle timeout")
if s.OnDisconnected != nil {
s.OnDisconnected()
}
return
}
switch msg.Type {
@@ -206,7 +239,16 @@ func (s *Session) Join(ctx context.Context, roomName string, trustedPeerID strin
if trustedPeerID != "" && msg.ID != trustedPeerID {
continue
}
log.Printf("flit: peer-join: %s", msg.ID[:8])
_ = s.connectTo(ctx, msg.ID)
case "peer-left":
if trustedPeerID != "" && msg.ID != trustedPeerID {
continue
}
// Log but don't reset: active file transfers use the WebRTC
// path which is independent of signaling. The stale-PC case
// is handled in connectTo when peer-join arrives next.
log.Printf("flit: peer-left: %s", msg.ID[:8])
case "from":
if trustedPeerID != "" && msg.From != trustedPeerID {
continue
@@ -220,45 +262,77 @@ func (s *Session) Join(ctx context.Context, roomName string, trustedPeerID strin
func (s *Session) connectTo(ctx context.Context, peerID string) error {
s.mu.Lock()
var oldPC *webrtc.PeerConnection
if s.pc != nil {
s.mu.Unlock()
return nil
state := s.pc.ConnectionState()
if state == webrtc.PeerConnectionStateNew ||
state == webrtc.PeerConnectionStateConnected {
s.mu.Unlock()
return nil // healthy or brand-new connection already exists
}
// Stale PC (peer left while we were connecting, or ICE failed) — replace it.
log.Printf("flit: connectTo %s: replacing stale PC (state=%s)", peerID[:8], state)
oldPC = s.pc
s.pc = nil
}
s.peerID = peerID
pc, err := webrtc.NewPeerConnection(webrtc.Configuration{ICEServers: iceServers(s.cfg)})
var se webrtc.SettingEngine
se.DetachDataChannels()
api := webrtc.NewAPI(webrtc.WithSettingEngine(se))
pc, err := api.NewPeerConnection(webrtc.Configuration{ICEServers: iceServers(s.cfg)})
if err != nil {
s.mu.Unlock()
if oldPC != nil {
_ = oldPC.Close()
}
return err
}
s.pc = pc
kp, err := crypto.GenerateEphemeral()
if err != nil {
s.pc = nil
s.mu.Unlock()
if oldPC != nil {
_ = oldPC.Close()
}
return err
}
s.esk, s.epk = kp.PrivateRaw(), kp.PublicRaw()
s.ekeySent = false
s.offered = false
s.peerEPK = nil
s.offerByOrder = s.identity.PeerID() < peerID
s.mu.Unlock()
// Close the stale PC after s.pc has been updated. The Closed state callback
// will see isCurrent=false and skip OnDisconnected, so runPeerLoop keeps running.
if oldPC != nil {
_ = oldPC.Close()
}
pc.OnDataChannel(func(dc *webrtc.DataChannel) { s.wireDC(dc) })
pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
if state == webrtc.PeerConnectionStateFailed ||
state == webrtc.PeerConnectionStateDisconnected ||
state == webrtc.PeerConnectionStateClosed {
s.mu.Lock()
s.pc = nil
s.ekeySent = false
s.offered = false
s.peerEPK = nil
isCurrent := s.pc == pc
if isCurrent {
s.pc = nil
s.ekeySent = false
s.offered = false
s.peerEPK = nil
}
s.mu.Unlock()
if s.OnDisconnected != nil {
log.Printf("flit: PC state=%s isCurrent=%v", state, isCurrent)
if isCurrent && s.OnDisconnected != nil {
s.OnDisconnected()
}
}
})
s.sendEkey()
offerByOrder := s.identity.PeerID() < peerID
if offerByOrder {
if s.offerByOrder {
go func() {
time.Sleep(ekeyTimeout)
s.maybeOffer()
@@ -267,6 +341,23 @@ func (s *Session) connectTo(ctx context.Context, peerID string) error {
return nil
}
// resetPeer tears down the current WebRTC connection without triggering
// OnDisconnected, leaving the signaling connection open so that the next
// peer-join for this peer can reconnect immediately.
func (s *Session) resetPeer() {
s.mu.Lock()
pc := s.pc
s.pc = nil
s.ekeySent = false
s.offered = false
s.peerEPK = nil
s.mu.Unlock()
if pc != nil {
log.Printf("flit: resetPeer: closing stale PC (peer left signaling)")
_ = pc.Close()
}
}
func (s *Session) sendEkey() {
s.mu.Lock()
if s.ekeySent {
@@ -290,7 +381,7 @@ func (s *Session) sendEkey() {
func (s *Session) maybeOffer() {
s.mu.Lock()
if s.offered || s.pc == nil {
if !s.offerByOrder || s.offered || s.pc == nil {
s.mu.Unlock()
return
}
@@ -434,8 +525,32 @@ func (s *Session) wireDC(dc *webrtc.DataChannel) {
s.mu.Lock()
s.dc = dc
s.mu.Unlock()
dc.OnOpen(func() { s.sendHello() })
dc.OnMessage(func(msg webrtc.DataChannelMessage) { s.onControl(msg.Data) })
startYaw := func() {
raw, err := dc.Detach()
if err != nil {
log.Printf("flit: yaw detach: %v", err)
return
}
s.sendHello()
go func() {
buf := make([]byte, 32768)
for {
n, err := raw.Read(buf)
if n > 0 {
s.onControl(append([]byte(nil), buf[:n]...))
}
if err != nil {
return
}
}
}()
}
if dc.ReadyState() == webrtc.DataChannelStateOpen {
startYaw()
} else {
dc.OnOpen(func() { startYaw() })
}
return
}
if strings.HasPrefix(dc.Label(), "f:") {
@@ -468,6 +583,19 @@ func (s *Session) onControl(data []byte) {
size, _ := m["size"].(float64)
s.OnFileOffer(FileOffer{XID: m["xid"].(string), Name: m["name"].(string), Size: int64(size)})
}
case "file-accept":
xid, _ := m["xid"].(string)
var offset int64
if v, ok := m["offset"].(string); ok {
offset, _ = strconv.ParseInt(v, 10, 64)
}
s.mu.Lock()
ch := s.pendingAccepts[xid]
delete(s.pendingAccepts, xid)
s.mu.Unlock()
if ch != nil {
ch <- offset
}
}
}
@@ -482,7 +610,9 @@ func (s *Session) controlSend(obj map[string]string) {
_ = dc.SendText(string(data))
}
// SendFile offers and streams a file to the connected peer.
// SendFile offers and streams a file to the connected peer. If the receiver
// has a partial .part file it will reply with a non-zero offset and the send
// resumes from that byte position.
func (s *Session) SendFile(path string) error {
f, err := os.Open(path)
if err != nil {
@@ -496,6 +626,14 @@ func (s *Session) SendFile(path string) error {
xid := fmt.Sprintf("push-%d", time.Now().UnixNano())
name := st.Name()
acceptCh := make(chan int64, 1)
s.mu.Lock()
if s.pendingAccepts == nil {
s.pendingAccepts = make(map[string]chan int64)
}
s.pendingAccepts[xid] = acceptCh
s.mu.Unlock()
s.controlSend(map[string]string{"type": "file-offer", "name": name, "size": fmt.Sprint(st.Size()), "xid": xid})
s.mu.Lock()
@@ -507,10 +645,24 @@ func (s *Session) SendFile(path string) error {
}
opened := make(chan struct{})
dc.OnOpen(func() { close(opened) })
var offset int64
select {
case offset = <-acceptCh:
case <-time.After(30 * time.Second):
return fmt.Errorf("timed out waiting for file-accept")
}
select {
case <-opened:
case <-time.After(15 * time.Second):
return fmt.Errorf("timed out waiting for file-accept")
return fmt.Errorf("timed out waiting for data channel open")
}
if offset > 0 {
if _, err := f.Seek(offset, io.SeekStart); err != nil {
return fmt.Errorf("seek to resume offset: %w", err)
}
log.Printf("flit: resuming %s from byte %d", name, offset)
}
buf := make([]byte, chunkSize)
@@ -534,11 +686,22 @@ func (s *Session) SendFile(path string) error {
return dc.Close()
}
// AcceptOffer accepts an incoming file-offer and writes the transfer to downloadDir.
// AcceptOffer accepts an incoming file-offer and writes the transfer to
// downloadDir. If a .part file from a previous partial transfer exists for this
// filename and is smaller than the declared size, its byte count is sent back
// as the resume offset so the sender can skip ahead.
func (s *Session) AcceptOffer(offer FileOffer, downloadDir string) {
s.controlSend(map[string]string{"type": "file-accept", "xid": offer.XID})
partPath := filepath.Join(downloadDir, offer.Name+".part")
var offset int64
if info, err := os.Stat(partPath); err == nil && info.Size() < offer.Size {
offset = info.Size()
}
s.controlSend(map[string]string{"type": "file-accept", "xid": offer.XID, "offset": fmt.Sprint(offset)})
s.mu.Lock()
s.pendingRecv = &recvState{offer: offer, dir: downloadDir}
if s.pendingRecvs == nil {
s.pendingRecvs = make(map[string]*recvState)
}
s.pendingRecvs[offer.XID] = &recvState{offer: offer, dir: downloadDir, have: offset}
s.mu.Unlock()
}
@@ -556,35 +719,87 @@ type recvState struct {
func (s *Session) wireFileRecv(dc *webrtc.DataChannel) {
xid := strings.TrimPrefix(dc.Label(), "f:")
s.mu.Lock()
rs := s.pendingRecv
s.pendingRecv = nil
rs := s.pendingRecvs[xid]
delete(s.pendingRecvs, xid)
s.mu.Unlock()
if rs == nil || rs.offer.XID != xid {
if rs == nil {
return
}
path := rs.dir + "/" + rs.offer.Name
f, err := os.Create(path)
if err != nil {
log.Printf("flit: create %s: %v", path, err)
return
}
rs.f = f
dc.OnMessage(func(msg webrtc.DataChannelMessage) {
if _, err := f.Write(msg.Data); err != nil {
log.Printf("flit: wireFileRecv: %s size=%d state=%s", rs.offer.Name, rs.offer.Size, dc.ReadyState())
// Use pion's Detach API so data is buffered in pion's SCTP stream and
// read by our goroutine — no OnMessage/OnClose callback race.
startRecv := func() {
raw, err := dc.Detach()
if err != nil {
log.Printf("flit: file DC detach %s: %v", rs.offer.Name, err)
return
}
rs.have += int64(len(msg.Data))
if s.OnFileProgress != nil {
s.OnFileProgress(xid, rs.have, rs.offer.Size)
}
})
dc.OnClose(func() {
f.Close()
if s.OnFileDone != nil {
s.OnFileDone(rs.offer.Name, path)
}
})
log.Printf("flit: file DC open: %s", rs.offer.Name)
go func() {
finalPath := filepath.Join(rs.dir, rs.offer.Name)
resuming := rs.have > 0
var f *os.File
if resuming {
partPath := finalPath + ".part"
f, err = os.OpenFile(partPath, os.O_WRONLY|os.O_APPEND, 0644)
if err != nil {
log.Printf("flit: reopen %s: %v — restarting", partPath, err)
resuming = false
rs.have = 0
f, err = os.Create(finalPath)
} else {
log.Printf("flit: resuming %s from byte %d", rs.offer.Name, rs.have)
}
} else {
f, err = os.Create(finalPath)
}
if err != nil {
log.Printf("flit: open output: %v", err)
return
}
buf := make([]byte, chunkSize)
for {
n, readErr := raw.Read(buf)
if n > 0 {
if _, werr := f.Write(buf[:n]); werr != nil {
log.Printf("flit: write %s: %v", rs.offer.Name, werr)
break
}
rs.have += int64(n)
if s.OnFileProgress != nil {
s.OnFileProgress(xid, rs.have, rs.offer.Size)
}
}
if readErr != nil {
break
}
}
f.Close()
log.Printf("flit: recv done: %s wrote=%d of %d", rs.offer.Name, rs.have, rs.offer.Size)
if resuming {
partPath := finalPath + ".part"
if rerr := os.Rename(partPath, finalPath); rerr != nil {
log.Printf("flit: rename %s: %v", partPath, rerr)
return
}
}
if s.OnFileDone != nil {
s.OnFileDone(rs.offer.Name, finalPath)
}
}()
}
if dc.ReadyState() == webrtc.DataChannelStateOpen {
startRecv()
} else {
dc.OnOpen(func() { startRecv() })
}
}
func mustHex(s string) []byte {