feat: implement §9 file transfer over dedicated binary DataChannels
File chunks travel over a per-transfer "f:<xid>" WebRTC DataChannel —
direct DTLS-encrypted P2P, the anchor never sees file data. Once the
initial handshake is done the anchor can disappear and transfers continue.
Key design choices:
- Receiver sends "ok" on the file DC before sender streams, eliminating
a race where tiny files could be fully sent/closed before the receiver's
OnMessage handler is registered (open-race §6 analogue for data DCs).
- Auto-accept: receiver accepts every incoming offer immediately.
- Download dir: per-network at <data-dir>/downloads-<netid>/.
- Backpressure: bufferedAmountLowThreshold to avoid overwhelming the DC.
- SHA-256 verified on receive; mismatches emit EvtError and discard temp file.
- IPC: send_file {peer_id, path} → offers the named file from share dir.
- EvtFileComplete {transfer_id, path} emitted on success.
IPC command: {"type":"send_file","peer_id":"<hex>","path":"<filename>"}
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Fredrik Johansson
@@ -6,6 +6,8 @@ import (
|
||||
"os"
|
||||
"sync"
|
||||
|
||||
"github.com/pion/webrtc/v3"
|
||||
|
||||
"github.com/waste-go/internal/crypto"
|
||||
"github.com/waste-go/internal/proto"
|
||||
"github.com/waste-go/internal/store"
|
||||
@@ -16,18 +18,26 @@ type PeerConn struct {
|
||||
Info proto.PeerInfo
|
||||
// Send a line of JSON to this peer (pre-encrypted by the sender goroutine).
|
||||
Send chan<- []byte
|
||||
// PC is the underlying PeerConnection, used to open additional DataChannels.
|
||||
PC *webrtc.PeerConnection
|
||||
}
|
||||
|
||||
// Mesh is the shared state of the local node.
|
||||
// All methods are safe to call from multiple goroutines.
|
||||
type Mesh struct {
|
||||
Identity *crypto.Identity
|
||||
Store *store.Store // may be nil if persistence is disabled
|
||||
ShareDir string // directory whose contents are shared with peers; "" = no sharing
|
||||
Identity *crypto.Identity
|
||||
Store *store.Store // may be nil if persistence is disabled
|
||||
ShareDir string // directory whose contents are shared with peers; "" = no sharing
|
||||
DownloadDir string // directory where received files are saved
|
||||
|
||||
mu sync.RWMutex
|
||||
peers map[proto.PeerID]*PeerConn
|
||||
|
||||
// file transfer state
|
||||
transferMu sync.Mutex
|
||||
outbound map[string]*outboundTransfer // xid → pending outbound
|
||||
inbound map[string]*inboundTransfer // xid → pending inbound
|
||||
|
||||
// subscribers receive a copy of every event (fan-out to IPC clients)
|
||||
subMu sync.Mutex
|
||||
subs []chan proto.IpcMessage
|
||||
@@ -40,6 +50,8 @@ func New(id *crypto.Identity, st *store.Store) *Mesh {
|
||||
Identity: id,
|
||||
Store: st,
|
||||
peers: make(map[proto.PeerID]*PeerConn),
|
||||
outbound: make(map[string]*outboundTransfer),
|
||||
inbound: make(map[string]*inboundTransfer),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -57,6 +57,7 @@ func WireDataChannel(
|
||||
peerConn := &PeerConn{
|
||||
Info: proto.PeerInfo{ID: peerID, Alias: string(peerID.Short())},
|
||||
Send: sendCh,
|
||||
PC: pc,
|
||||
}
|
||||
m.AddPeer(peerConn)
|
||||
|
||||
@@ -211,6 +212,16 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
|
||||
PeerID: peerIDPtr(from),
|
||||
Offer: &proto.FileOffer{Xid: msg.Xid, Name: msg.Name, Size: msg.Size, SHA256: msg.SHA256},
|
||||
})
|
||||
m.acceptIncoming(msg, from)
|
||||
|
||||
case proto.MsgFileAccept:
|
||||
m.startSend(msg.Xid, from)
|
||||
|
||||
case proto.MsgFileCancel:
|
||||
log.Printf("mesh: file-cancel from %s xid=%s reason=%s", from.Short(), msg.Xid, msg.Reason)
|
||||
|
||||
case proto.MsgFileDone:
|
||||
log.Printf("mesh: file-done from %s xid=%s", from.Short(), msg.Xid)
|
||||
|
||||
case proto.MsgPeerGossip:
|
||||
if msg.Gossip != nil {
|
||||
|
||||
351
internal/mesh/transfer.go
Normal file
351
internal/mesh/transfer.go
Normal file
@@ -0,0 +1,351 @@
|
||||
package mesh
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hash"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/pion/webrtc/v3"
|
||||
|
||||
"github.com/waste-go/internal/proto"
|
||||
)
|
||||
|
||||
const (
|
||||
fileChunkSize = 64 * 1024 // 64 KiB per chunk
|
||||
fileBufferHighWater = 512 * 1024 // pause sending above this
|
||||
fileBufferLowWater = 256 * 1024 // resume when it drops here
|
||||
)
|
||||
|
||||
type outboundTransfer struct {
|
||||
peerID proto.PeerID
|
||||
path string
|
||||
size int64
|
||||
sha256 string
|
||||
}
|
||||
|
||||
type inboundTransfer struct {
|
||||
from proto.PeerID
|
||||
name string
|
||||
size int64
|
||||
sha256 string
|
||||
mu sync.Mutex
|
||||
tmp *os.File
|
||||
hasher hash.Hash
|
||||
written int64
|
||||
}
|
||||
|
||||
// OfferFile reads filename from ShareDir, computes its SHA-256, and sends a
|
||||
// file-offer to peerID over the existing "yaw" DataChannel.
|
||||
func (m *Mesh) OfferFile(peerID proto.PeerID, filename string) error {
|
||||
if m.ShareDir == "" {
|
||||
return fmt.Errorf("no share directory configured")
|
||||
}
|
||||
if strings.ContainsAny(filename, "/\\") || filename == ".." {
|
||||
return fmt.Errorf("invalid filename %q", filename)
|
||||
}
|
||||
path := filepath.Join(m.ShareDir, filename)
|
||||
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open %q: %w", filename, err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
h := sha256.New()
|
||||
if _, err := io.Copy(h, f); err != nil {
|
||||
return fmt.Errorf("hash %q: %w", filename, err)
|
||||
}
|
||||
shaHex := hex.EncodeToString(h.Sum(nil))
|
||||
xid := newXid()
|
||||
|
||||
m.transferMu.Lock()
|
||||
m.outbound[xid] = &outboundTransfer{
|
||||
peerID: peerID,
|
||||
path: path,
|
||||
size: info.Size(),
|
||||
sha256: shaHex,
|
||||
}
|
||||
m.transferMu.Unlock()
|
||||
|
||||
wire, err := json.Marshal(proto.PeerMessage{
|
||||
Type: proto.MsgFileOffer,
|
||||
Xid: xid,
|
||||
Name: filename,
|
||||
Size: info.Size(),
|
||||
SHA256: shaHex,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !m.SendTo(peerID, wire) {
|
||||
m.transferMu.Lock()
|
||||
delete(m.outbound, xid)
|
||||
m.transferMu.Unlock()
|
||||
return fmt.Errorf("peer %s not connected", peerID.Short())
|
||||
}
|
||||
log.Printf("transfer: offered %s (%d bytes) to %s xid=%s", filename, info.Size(), peerID.Short(), xid[:8])
|
||||
return nil
|
||||
}
|
||||
|
||||
// acceptIncoming is called when we receive file-offer from a peer.
|
||||
// It stores the inbound state and immediately sends file-accept (auto-accept).
|
||||
func (m *Mesh) acceptIncoming(msg proto.PeerMessage, from proto.PeerID) {
|
||||
m.transferMu.Lock()
|
||||
m.inbound[msg.Xid] = &inboundTransfer{
|
||||
from: from,
|
||||
name: msg.Name,
|
||||
size: msg.Size,
|
||||
sha256: msg.SHA256,
|
||||
}
|
||||
m.transferMu.Unlock()
|
||||
|
||||
accept, _ := json.Marshal(proto.PeerMessage{
|
||||
Type: proto.MsgFileAccept,
|
||||
Xid: msg.Xid,
|
||||
})
|
||||
m.SendTo(from, accept)
|
||||
log.Printf("transfer: auto-accepted %s (%d bytes) from %s xid=%s", msg.Name, msg.Size, from.Short(), msg.Xid[:8])
|
||||
}
|
||||
|
||||
// startSend opens the binary DataChannel "f:<xid>" and streams the file.
|
||||
// Called when we receive file-accept from the peer.
|
||||
func (m *Mesh) startSend(xid string, from proto.PeerID) {
|
||||
m.transferMu.Lock()
|
||||
t, ok := m.outbound[xid]
|
||||
m.transferMu.Unlock()
|
||||
if !ok {
|
||||
log.Printf("transfer: file-accept for unknown xid %s", xid[:8])
|
||||
return
|
||||
}
|
||||
if t.peerID != from {
|
||||
log.Printf("transfer: file-accept from unexpected peer %s", from.Short())
|
||||
return
|
||||
}
|
||||
|
||||
m.mu.RLock()
|
||||
conn, ok := m.peers[from]
|
||||
m.mu.RUnlock()
|
||||
if !ok {
|
||||
log.Printf("transfer: peer %s not connected for xid %s", from.Short(), xid[:8])
|
||||
return
|
||||
}
|
||||
|
||||
ordered := true
|
||||
dc, err := conn.PC.CreateDataChannel("f:"+xid, &webrtc.DataChannelInit{Ordered: &ordered})
|
||||
if err != nil {
|
||||
log.Printf("transfer: create file DC xid=%s: %v", xid[:8], err)
|
||||
return
|
||||
}
|
||||
|
||||
// Wait for receiver to signal "ok" before sending — this ensures the receiver
|
||||
// has its OnMessage/OnClose handlers registered before any data arrives.
|
||||
dc.OnOpen(func() {
|
||||
dc.OnMessage(func(msg webrtc.DataChannelMessage) {
|
||||
if msg.IsString && string(msg.Data) == "ok" {
|
||||
go m.sendFileChunks(dc, t, xid)
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid string) {
|
||||
defer func() {
|
||||
m.transferMu.Lock()
|
||||
delete(m.outbound, xid)
|
||||
m.transferMu.Unlock()
|
||||
}()
|
||||
|
||||
f, err := os.Open(t.path)
|
||||
if err != nil {
|
||||
log.Printf("transfer: open file xid=%s: %v", xid[:8], err)
|
||||
dc.Close()
|
||||
return
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
// Backpressure: block when the DC buffer is full.
|
||||
resume := make(chan struct{}, 1)
|
||||
dc.SetBufferedAmountLowThreshold(fileBufferLowWater)
|
||||
dc.OnBufferedAmountLow(func() {
|
||||
select {
|
||||
case resume <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
})
|
||||
|
||||
buf := make([]byte, fileChunkSize)
|
||||
var sent int64
|
||||
|
||||
for {
|
||||
n, readErr := f.Read(buf)
|
||||
if n > 0 {
|
||||
for dc.BufferedAmount() > fileBufferHighWater {
|
||||
<-resume
|
||||
}
|
||||
if err := dc.Send(buf[:n]); err != nil {
|
||||
log.Printf("transfer: send chunk xid=%s: %v", xid[:8], err)
|
||||
return
|
||||
}
|
||||
sent += int64(n)
|
||||
m.Emit(proto.IpcMessage{
|
||||
Type: proto.EvtFileProgress,
|
||||
TransferID: xid,
|
||||
BytesReceived: sent,
|
||||
TotalBytes: t.size,
|
||||
})
|
||||
}
|
||||
if readErr == io.EOF {
|
||||
break
|
||||
}
|
||||
if readErr != nil {
|
||||
log.Printf("transfer: read file xid=%s: %v", xid[:8], readErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
dc.Close()
|
||||
|
||||
done, _ := json.Marshal(proto.PeerMessage{
|
||||
Type: proto.MsgFileDone,
|
||||
Xid: xid,
|
||||
SHA256: t.sha256,
|
||||
})
|
||||
m.SendTo(t.peerID, done)
|
||||
log.Printf("transfer: sent %s (%d bytes) xid=%s", filepath.Base(t.path), sent, xid[:8])
|
||||
}
|
||||
|
||||
// HandleInboundFileDC is called from the anchor when a "f:<xid>" DataChannel
|
||||
// arrives on a PeerConnection. It writes chunks to a temp file, verifies the
|
||||
// SHA-256 on close, and emits EvtFileComplete.
|
||||
func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from proto.PeerID) {
|
||||
m.transferMu.Lock()
|
||||
t, ok := m.inbound[xid]
|
||||
m.transferMu.Unlock()
|
||||
if !ok {
|
||||
log.Printf("transfer: no pending inbound for xid %s", xid[:8])
|
||||
dc.Close()
|
||||
return
|
||||
}
|
||||
|
||||
var once sync.Once
|
||||
doOpen := func() {
|
||||
if err := os.MkdirAll(m.DownloadDir, 0o755); err != nil {
|
||||
log.Printf("transfer: mkdir %s: %v", m.DownloadDir, err)
|
||||
return
|
||||
}
|
||||
tmp, err := os.CreateTemp(m.DownloadDir, "dl-*.tmp")
|
||||
if err != nil {
|
||||
log.Printf("transfer: create temp file: %v", err)
|
||||
return
|
||||
}
|
||||
t.mu.Lock()
|
||||
t.tmp = tmp
|
||||
t.hasher = sha256.New()
|
||||
t.mu.Unlock()
|
||||
log.Printf("transfer: receiving %s xid=%s", t.name, xid[:8])
|
||||
// Signal sender that we are ready — it will not start streaming until
|
||||
// it receives this, guaranteeing our OnMessage/OnClose are registered first.
|
||||
dc.SendText("ok") //nolint:errcheck
|
||||
}
|
||||
|
||||
// Register OnMessage first so pion queues any early binary chunks.
|
||||
dc.OnMessage(func(msg webrtc.DataChannelMessage) {
|
||||
if msg.IsString {
|
||||
return // "ok" echo or any other text; ignore
|
||||
}
|
||||
t.mu.Lock()
|
||||
tmp, h := t.tmp, t.hasher
|
||||
t.mu.Unlock()
|
||||
if tmp == nil {
|
||||
return
|
||||
}
|
||||
if _, err := tmp.Write(msg.Data); err != nil {
|
||||
log.Printf("transfer: write xid=%s: %v", xid[:8], err)
|
||||
return
|
||||
}
|
||||
h.Write(msg.Data)
|
||||
t.mu.Lock()
|
||||
t.written += int64(len(msg.Data))
|
||||
written := t.written
|
||||
t.mu.Unlock()
|
||||
m.Emit(proto.IpcMessage{
|
||||
Type: proto.EvtFileProgress,
|
||||
TransferID: xid,
|
||||
BytesReceived: written,
|
||||
TotalBytes: t.size,
|
||||
})
|
||||
})
|
||||
|
||||
dc.OnOpen(func() { once.Do(doOpen) })
|
||||
if dc.ReadyState() == webrtc.DataChannelStateOpen {
|
||||
once.Do(doOpen)
|
||||
}
|
||||
|
||||
dc.OnClose(func() {
|
||||
t.mu.Lock()
|
||||
tmp := t.tmp
|
||||
actualSha := ""
|
||||
if t.hasher != nil {
|
||||
actualSha = hex.EncodeToString(t.hasher.Sum(nil))
|
||||
}
|
||||
name, expectedSha := t.name, t.sha256
|
||||
t.mu.Unlock()
|
||||
|
||||
m.transferMu.Lock()
|
||||
delete(m.inbound, xid)
|
||||
m.transferMu.Unlock()
|
||||
|
||||
if tmp == nil {
|
||||
return
|
||||
}
|
||||
tmp.Close()
|
||||
|
||||
if actualSha != expectedSha {
|
||||
os.Remove(tmp.Name())
|
||||
log.Printf("transfer: sha256 mismatch xid=%s got=%s want=%s", xid[:8], actualSha[:8], expectedSha[:8])
|
||||
m.Emit(proto.IpcMessage{
|
||||
Type: proto.EvtError,
|
||||
TransferID: xid,
|
||||
ErrorMessage: fmt.Sprintf("file %s: sha256 mismatch", name),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
final := filepath.Join(m.DownloadDir, name)
|
||||
if _, err := os.Stat(final); err == nil {
|
||||
ext := filepath.Ext(name)
|
||||
base := strings.TrimSuffix(name, ext)
|
||||
final = filepath.Join(m.DownloadDir, base+"-"+xid[:8]+ext)
|
||||
}
|
||||
if err := os.Rename(tmp.Name(), final); err != nil {
|
||||
log.Printf("transfer: rename xid=%s: %v", xid[:8], err)
|
||||
return
|
||||
}
|
||||
log.Printf("transfer: saved %s -> %s", name, final)
|
||||
m.Emit(proto.IpcMessage{
|
||||
Type: proto.EvtFileComplete,
|
||||
TransferID: xid,
|
||||
Path: final,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func newXid() string {
|
||||
b := make([]byte, 16)
|
||||
rand.Read(b) //nolint:errcheck
|
||||
return hex.EncodeToString(b)
|
||||
}
|
||||
Reference in New Issue
Block a user