Add "share status" command and a relay-piped fallback for cross-network peers

status: swarm.Client.Status()/RoomStatus/FileStatus report per-room peer
count and per-file chunk progress; daemon gets a "status" control action;
CLI gets "share status" to display it.

Relay fallback: peers advertising a LAN-only address can't be dialed
directly by peers on a different network. connectToPeer now races a
direct dial against a relay-piped connection (both sides derive the same
pairing key from their two IDs, independent of who arrives first) and
uses whichever succeeds. The relay grows a second, on-demand path
alongside room membership: a "pipe" request pairs two connections on a
key and forwards raw bytes between them from then on, same as the
original relay's behavior before the rendezvous-only redesign, just now
triggered per-pair only when direct connection fails.
This commit is contained in:
2026-07-20 12:27:27 +08:00
parent 26cef8ee12
commit 15a6658d1c
7 changed files with 411 additions and 43 deletions
BIN
View File
Binary file not shown.
+42
View File
@@ -34,12 +34,16 @@ USAGE EXAMPLES:
Join a room to receive whatever's shared there (and help re-share it):
share secret-code
See what rooms are currently active:
share status
Stop the background daemon (leaves every room it's in):
share stop
COMMANDS:
send seed file(s) into a room (see options with: share send -h)
relay start your own relay (optional; see options with: share relay -h)
status show the rooms the background daemon is currently in
stop stop the background daemon and leave every room
daemon run the background daemon in the foreground (mostly for debugging)
@@ -77,6 +81,8 @@ func Run() error {
return runRelay(splice(args, i))
case "stop":
return runStop(splice(args, i))
case "status":
return runStatus(splice(args, i))
case "daemon":
return runDaemon(splice(args, i))
}
@@ -244,6 +250,42 @@ func runStop(args []string) error {
return nil
}
func runStatus(args []string) error {
fs := flag.NewFlagSet("status", flag.ExitOnError)
fs.Usage = func() {
fmt.Fprintln(os.Stderr, "share status")
}
if err := fs.Parse(args); err != nil {
return err
}
rooms, err := daemon.Status()
if err != nil {
fmt.Println(err)
return nil
}
if len(rooms) == 0 {
fmt.Println("No active rooms.")
return nil
}
for _, r := range rooms {
peerWord := "peers"
if r.Peers == 1 {
peerWord = "peer"
}
fmt.Printf("Room %s (%d %s)\n", r.Code, r.Peers, peerWord)
for _, f := range r.Files {
state := "in progress"
if f.Complete {
state = "complete"
}
fmt.Printf(" %-24s %10s %d/%d chunks %s\n", f.Name, utils.ByteCountDecimal(f.Size), f.HaveChunks, f.TotalChunks, state)
}
}
return nil
}
func runDaemon(args []string) error {
fs := flag.NewFlagSet("daemon", flag.ExitOnError)
fs.Usage = func() {
+26 -9
View File
@@ -8,6 +8,8 @@ import (
"os/exec"
"syscall"
"time"
"github.com/tzgyn/share/src/swarm"
)
// IsRunning reports whether a daemon is currently listening on the
@@ -71,7 +73,7 @@ func EnsureRunning(debug bool) error {
// and outDir must already be absolute paths — the daemon may have been
// started long ago from a different working directory.
func Join(relayAddress, code string, files []string, outDir string, oneTime bool) error {
return call(request{
_, err := call(request{
Action: "join",
Join: &joinRequest{
RelayAddress: relayAddress,
@@ -81,6 +83,7 @@ func Join(relayAddress, code string, files []string, outDir string, oneTime bool
OneTime: oneTime,
},
})
return err
}
// Stop tells the running daemon to leave every room it's in and exit.
@@ -88,29 +91,43 @@ func Stop() error {
if !IsRunning() {
return fmt.Errorf("daemon is not running")
}
return call(request{Action: "stop"})
_, err := call(request{Action: "stop"})
return err
}
func call(req request) error {
// Status asks the running daemon which rooms it's currently in and how
// each one's transfers are progressing.
func Status() ([]swarm.RoomStatus, error) {
if !IsRunning() {
return nil, fmt.Errorf("daemon is not running")
}
resp, err := call(request{Action: "status"})
if err != nil {
return nil, err
}
return resp.Rooms, nil
}
func call(req request) (response, error) {
path, err := socketPath()
if err != nil {
return err
return response{}, err
}
conn, err := net.DialTimeout("unix", path, 2*time.Second)
if err != nil {
return fmt.Errorf("could not reach daemon: %w", err)
return response{}, fmt.Errorf("could not reach daemon: %w", err)
}
defer conn.Close()
if err := json.NewEncoder(conn).Encode(req); err != nil {
return err
return response{}, err
}
var resp response
if err := json.NewDecoder(conn).Decode(&resp); err != nil {
return err
return response{}, err
}
if !resp.OK {
return fmt.Errorf("%s", resp.Error)
return resp, fmt.Errorf("%s", resp.Error)
}
return nil
return resp, nil
}
+6 -3
View File
@@ -1,9 +1,11 @@
package daemon
import "github.com/tzgyn/share/src/swarm"
// request/response are the only vocabulary spoken over the control
// socket between a CLI invocation and the background daemon.
type request struct {
Action string `json:"action"` // "join" or "stop"
Action string `json:"action"` // "join", "stop", or "status"
Join *joinRequest `json:"join,omitempty"`
}
@@ -20,6 +22,7 @@ type joinRequest struct {
}
type response struct {
OK bool `json:"ok"`
Error string `json:"error,omitempty"`
OK bool `json:"ok"`
Error string `json:"error,omitempty"`
Rooms []swarm.RoomStatus `json:"rooms,omitempty"` // "status" only
}
+13
View File
@@ -81,6 +81,9 @@ func (s *server) handleConn(conn net.Conn) (shutdown bool) {
case "stop":
respond(conn, response{OK: true})
return true
case "status":
respond(conn, response{OK: true, Rooms: s.status()})
return false
default:
respond(conn, response{OK: false, Error: fmt.Sprintf("unknown action %q", req.Action)})
return false
@@ -140,3 +143,13 @@ func (s *server) stopAll() {
cl.Stop()
}
}
func (s *server) status() []swarm.RoomStatus {
s.mu.Lock()
defer s.mu.Unlock()
statuses := make([]swarm.RoomStatus, 0, len(s.rooms))
for _, cl := range s.rooms {
statuses = append(statuses, cl.Status())
}
return statuses
}
+109 -14
View File
@@ -134,6 +134,44 @@ func (cl *Client) Wait() error {
return <-cl.done
}
// FileStatus summarizes one file's transfer progress within a room.
type FileStatus struct {
Name string
Size int64
HaveChunks int64
TotalChunks int64
Complete bool
}
// RoomStatus summarizes one room this client is participating in.
type RoomStatus struct {
Code string
Peers int
Files []FileStatus
}
// Status reports this room's current code, peer count, and per-file
// transfer progress.
func (cl *Client) Status() RoomStatus {
cl.mu.Lock()
defer cl.mu.Unlock()
files := make([]FileStatus, 0, len(cl.files))
for _, f := range cl.files {
files = append(files, FileStatus{
Name: f.entry.Name,
Size: f.entry.Size,
HaveChunks: int64(len(f.store.HaveOffsets())),
TotalChunks: NumChunks(f.entry.Size),
Complete: f.store.Complete(),
})
}
return RoomStatus{
Code: cl.opts.Code,
Peers: len(cl.links),
Files: files,
}
}
// Stop ends this room's participation. Wait will then return nil.
func (cl *Client) Stop() {
cl.stopOnce.Do(func() { close(cl.stopCh) })
@@ -154,7 +192,7 @@ func (cl *Client) loop() {
return
}
for _, p := range peers {
if p.ID == cl.selfID || p.Addr == "" {
if p.ID == cl.selfID {
continue
}
cl.dialedMu.Lock()
@@ -162,7 +200,7 @@ func (cl *Client) loop() {
cl.dialed[p.ID] = true
cl.dialedMu.Unlock()
if !already {
go cl.dialPeer(p)
go cl.connectToPeer(p)
}
}
case <-time.After(500 * time.Millisecond):
@@ -246,20 +284,77 @@ func (cl *Client) acceptLoop(l net.Listener) {
}
}
func (cl *Client) dialPeer(p tcp.PeerInfo) {
conn, err := net.DialTimeout("tcp", p.Addr, 5*time.Second)
if err != nil {
log.Debugf("could not dial peer %s at %s: %v", p.ID, p.Addr, err)
return
// connectToPeer races a direct dial against a relay-piped fallback for
// reaching p; whichever succeeds first is used, and the other attempt is
// abandoned. This is what makes peers on different networks (no shared
// LAN, no forwarded port) work at all: when direct fails, the relay just
// pipes bytes between the two of them instead — the same as croc's own
// default behavior for anyone not on the same local network.
func (cl *Client) connectToPeer(p tcp.PeerInfo) {
type dialResult struct {
conn *comm.Comm
key []byte
}
c := comm.New(conn)
key, err := handshake(c, cl.opts.Code, true)
if err != nil {
log.Debugf("handshake with peer %s failed: %v", p.ID, err)
conn.Close()
return
found := make(chan dialResult, 1)
win := func(conn *comm.Comm, key []byte) {
select {
case found <- dialResult{conn, key}:
default:
conn.Close()
}
}
cl.handlePeerLink(newPeerLink(c, key))
if p.Addr != "" {
go func() {
conn, err := net.DialTimeout("tcp", p.Addr, 5*time.Second)
if err != nil {
log.Debugf("could not dial peer %s directly at %s: %v", p.ID, p.Addr, err)
return
}
c := comm.New(conn)
key, err := handshake(c, cl.opts.Code, true)
if err != nil {
log.Debugf("handshake with peer %s failed: %v", p.ID, err)
conn.Close()
return
}
win(c, key)
}()
}
go func() {
c, err := tcp.RelayPipe(cl.opts.RelayAddress, pairKey(cl.opts.Code, cl.selfID, p.ID))
if err != nil {
log.Debugf("could not open a relay pipe to peer %s: %v", p.ID, err)
return
}
// Both sides independently compute the same initiator role from
// the same two IDs, so exactly one of them ends up true.
key, err := handshake(c, cl.opts.Code, cl.selfID < p.ID)
if err != nil {
log.Debugf("relay-piped handshake with peer %s failed: %v", p.ID, err)
c.Close()
return
}
win(c, key)
}()
select {
case r := <-found:
cl.handlePeerLink(newPeerLink(r.conn, r.key))
case <-time.After(25 * time.Second):
log.Warnf("could not connect to peer %s: direct dial and relay-piped fallback both failed", p.ID)
}
}
// pairKey deterministically derives the same relay pipe-pairing key for
// both sides of a peer pair, regardless of which one computes it first.
func pairKey(code, idA, idB string) string {
if idA > idB {
idA, idB = idB, idA
}
return code + ":" + idA + ":" + idB
}
func (cl *Client) handlePeerLink(link *peerLink) {
+215 -17
View File
@@ -12,8 +12,16 @@ import (
log "github.com/tzgyn/share/src/logger"
"github.com/tzgyn/share/src/comm"
"github.com/tzgyn/share/src/models"
)
// pipeWaitTimeout bounds how long the relay holds a lone "pipe" request
// open waiting for its matching partner before giving up. Kept short: a
// peer only opens one of these right after a direct connection attempt to
// another peer has already failed, so both sides should arrive at nearly
// the same time.
const pipeWaitTimeout = 20 * time.Second
// PeerInfo is what the relay tells other room members about one peer:
// an identifier and the address others can dial to reach it directly
// (empty if this peer can't accept incoming connections, e.g. behind NAT).
@@ -22,13 +30,21 @@ type PeerInfo struct {
Addr string `json:"addr"`
}
// ControlMessage is the entire vocabulary spoken between a peer and the
// relay. A peer sends exactly one "join" to announce itself; the relay
// ControlMessage is the vocabulary spoken between a peer and the relay.
// A peer sends exactly one "join" to announce itself in a room; the relay
// pushes a "peers" message with the current room membership whenever it
// changes. The relay never sees file content or chunk data at all — peers
// exchange that directly with each other once they have each other's Addr.
// changes. The relay never sees file content or chunk data for that —
// peers exchange it directly with each other once they have each other's
// Addr.
//
// Separately, a peer can send a "pipe" instead of "join": this is for
// when two peers couldn't reach each other directly (different networks,
// no forwarded port). Two peers sending "pipe" with the same Room value
// (used here as an arbitrary pairing key, not an actual room) get piped
// together — the relay just forwards raw bytes between them from that
// point on, blind to whatever protocol the two peers speak over it.
type ControlMessage struct {
Type string `json:"type"` // "join" or "peers"
Type string `json:"type"` // "join", "peers", or "pipe"
Room string `json:"room,omitempty"`
Peer PeerInfo `json:"peer,omitempty"`
Peers []PeerInfo `json:"peers,omitempty"`
@@ -39,6 +55,7 @@ type server struct {
port string
debugLevel string
rooms roomMap
pipes pipeMap
*stop
}
@@ -57,6 +74,17 @@ type roomMap struct {
sync.Mutex
}
// pipeSlot is a lone "pipe" request waiting for its matching partner.
type pipeSlot struct {
comm *comm.Comm
ready chan *comm.Comm // the partner's comm, posted here once it arrives
}
type pipeMap struct {
slots map[string]*pipeSlot
sync.Mutex
}
func newDefaultServer() *server {
s := new(server)
s.debugLevel = DEFAULT_LOG_LEVEL
@@ -89,6 +117,10 @@ func (s *server) start() (err error) {
s.rooms.rooms = make(map[string]*roomInfo)
s.rooms.Unlock()
s.pipes.Lock()
s.pipes.slots = make(map[string]*pipeSlot)
s.pipes.Unlock()
defer s.stop.Cancel()
if s.stop.gui {
defer s.stop.wg.Wait()
@@ -149,7 +181,7 @@ func (s *server) run() (err error) {
if err != nil {
return fmt.Errorf("problem accepting connection: %w", err)
}
log.Debugf("client %s connected", connection.RemoteAddr().String())
log.Infof("client %s connected", connection.RemoteAddr().String())
s.stop.wg.Add(1)
go func(connection net.Conn) {
defer s.stop.wg.Done()
@@ -158,25 +190,59 @@ func (s *server) run() (err error) {
}
}
// handleConnection is the entire relay protocol: read one join message,
// register the peer in its room, broadcast membership, then just wait for
// the connection to close so we know when to remove the peer again. No
// data ever flows through here beyond membership control messages.
// handleConnection reads one control message and dispatches on its type:
// "join" registers the peer in a room and holds the connection open for
// membership control messages; "pipe" pairs this connection with another
// peer's matching "pipe" request and forwards raw bytes between them —
// the fallback for when two peers couldn't reach each other directly.
func (s *server) handleConnection(connection net.Conn) {
c := comm.New(connection)
defer connection.Close()
b, err := c.Receive()
if err != nil {
log.Debug(err)
// includes the relay's own startup self-check connection, which
// dials and closes without sending anything — routine, not a
// sign of a real client failing to join.
log.Infof("client %s disconnected before sending a control message: %v", connection.RemoteAddr().String(), err)
connection.Close()
return
}
var msg ControlMessage
if err := json.Unmarshal(b, &msg); err != nil || msg.Type != "join" || msg.Room == "" || msg.Peer.ID == "" {
log.Debugf("bad join message: %v", err)
if err := json.Unmarshal(b, &msg); err != nil {
log.Warnf("client %s sent an unparseable control message: %v", connection.RemoteAddr().String(), err)
connection.Close()
return
}
log.Debugf("peer %s joining room %s", msg.Peer.ID, msg.Room)
switch msg.Type {
case "join":
if msg.Room == "" || msg.Peer.ID == "" {
log.Warnf("client %s sent a bad join message (room=%q peerID=%q)", connection.RemoteAddr().String(), msg.Room, msg.Peer.ID)
connection.Close()
return
}
s.handleJoin(c, connection, msg)
case "pipe":
if msg.Room == "" {
log.Warnf("client %s sent a bad pipe request (empty key)", connection.RemoteAddr().String())
connection.Close()
return
}
s.handlePipe(c, msg.Room)
default:
log.Warnf("client %s sent an unknown control message type %q", connection.RemoteAddr().String(), msg.Type)
connection.Close()
}
}
// handleJoin registers the peer in its room, broadcasts membership, then
// just waits for the connection to close so we know when to remove the
// peer again. No data ever flows through here beyond membership control
// messages.
func (s *server) handleJoin(c *comm.Comm, connection net.Conn, msg ControlMessage) {
defer connection.Close()
log.Infof("peer %s (%s) joining room %s", msg.Peer.ID, connection.RemoteAddr().String(), msg.Room)
s.rooms.Lock()
r, ok := s.rooms.rooms[msg.Room]
@@ -203,12 +269,105 @@ func (s *server) handleConnection(connection net.Conn) {
// nothing more to read from a peer; block here until it disconnects
for {
if _, err := c.Receive(); err != nil {
log.Debugf("peer %s left room %s: %v", msg.Peer.ID, msg.Room, err)
log.Infof("peer %s left room %s: %v", msg.Peer.ID, msg.Room, err)
return
}
}
}
// handlePipe pairs this connection with another peer's "pipe" request
// carrying the same key, then forwards raw bytes between them until
// either side disconnects. The relay never parses anything past this
// point — whatever protocol the two peers speak over the pipe is
// invisible to it, same as a direct connection between them would be.
func (s *server) handlePipe(c *comm.Comm, key string) {
s.pipes.Lock()
slot, waiting := s.pipes.slots[key]
if !waiting {
slot = &pipeSlot{comm: c, ready: make(chan *comm.Comm, 1)}
s.pipes.slots[key] = slot
s.pipes.Unlock()
log.Infof("pipe %s: waiting for a match", key)
select {
case partner := <-slot.ready:
log.Infof("pipe %s: matched, forwarding bytes", key)
pipe(c.Connection(), partner.Connection())
c.Close()
partner.Close()
log.Infof("pipe %s: finished", key)
case <-time.After(pipeWaitTimeout):
s.pipes.Lock()
if s.pipes.slots[key] == slot {
delete(s.pipes.slots, key)
}
s.pipes.Unlock()
log.Infof("pipe %s: timed out waiting for a match", key)
c.Close()
}
return
}
delete(s.pipes.slots, key)
s.pipes.Unlock()
// hand off to the waiting goroutine above, which owns closing both
// connections once piping finishes — this one must not close its own.
slot.ready <- c
}
// chanFromConn creates a channel from a Conn object, and sends everything
// it Read()s from the socket to the channel.
func chanFromConn(conn net.Conn) chan []byte {
c := make(chan []byte, 1)
if err := conn.SetReadDeadline(time.Now().Add(3 * time.Hour)); err != nil {
log.Warnf("can't set read deadline: %v", err)
}
go func() {
b := make([]byte, models.TCP_BUFFER_SIZE)
for {
n, err := conn.Read(b)
if n > 0 {
res := make([]byte, n)
copy(res, b[:n])
c <- res
}
if err != nil {
c <- nil
break
}
}
}()
return c
}
// pipe creates a full-duplex pipe between the two sockets, transferring
// data from one to the other until either side closes.
func pipe(conn1, conn2 net.Conn) {
chan1 := chanFromConn(conn1)
chan2 := chanFromConn(conn2)
for {
select {
case b1 := <-chan1:
if b1 == nil {
return
}
if _, err := conn2.Write(b1); err != nil {
log.Warnf("pipe write error: %v", err)
}
case b2 := <-chan2:
if b2 == nil {
return
}
if _, err := conn1.Write(b2); err != nil {
log.Warnf("pipe write error: %v", err)
}
}
}
}
func (s *server) broadcastPeers(room string) {
s.rooms.Lock()
r, ok := s.rooms.rooms[room]
@@ -224,9 +383,19 @@ func (s *server) broadcastPeers(room string) {
}
s.rooms.Unlock()
if !ok {
log.Infof("room %s is now empty", room)
return
}
addrs := make([]string, len(peers))
for i, p := range peers {
addrs[i] = p.Addr
if addrs[i] == "" {
addrs[i] = "(unreachable/no listener)"
}
}
log.Infof("room %s now has %d peer(s): %v", room, len(peers), addrs)
b, err := json.Marshal(ControlMessage{Type: "peers", Peers: peers})
if err != nil {
log.Error(err)
@@ -234,7 +403,7 @@ func (s *server) broadcastPeers(room string) {
}
for _, c := range conns {
if err := c.Send(b); err != nil {
log.Debug(err)
log.Warnf("could not send peer list to %s: %v", c.Connection().RemoteAddr().String(), err)
}
}
}
@@ -287,3 +456,32 @@ func JoinRoom(address, room string, self PeerInfo, timelimit ...time.Duration) (
peers = ch
return
}
// RelayPipe connects to the relay and asks it to pipe this connection
// together with whichever other peer sends the same pairKey — the
// fallback for when two peers can't reach each other directly (different
// networks, no forwarded port, etc). pairKey must be something both
// peers compute identically and independently (e.g. derived from both
// their IDs, sorted, so it doesn't matter who arrives first). The
// returned connection can be used exactly like a direct one — the relay
// forwards whatever either side writes to it verbatim once both sides
// have shown up.
func RelayPipe(address, pairKey string, timelimit ...time.Duration) (c *comm.Comm, err error) {
if len(timelimit) > 0 {
c, err = comm.NewConnection(address, timelimit[0])
} else {
c, err = comm.NewConnection(address)
}
if err != nil {
return nil, err
}
b, err := json.Marshal(ControlMessage{Type: "pipe", Room: pairKey})
if err != nil {
return nil, err
}
if err := c.Send(b); err != nil {
return nil, err
}
return c, nil
}