Persist rooms across daemon restarts; add "share restart"; fix peer

count and a startup hang

- src/daemon/persist.go: rooms.json records every active room so a
  restarted daemon (crash, reboot, or "share stop" then relaunch) resumes
  them automatically. A --one-time room clears itself once it finishes
  on its own; anything else stays persisted since it only ever stops via
  an explicit Stop.
- "share restart": stops the running daemon, waits for it to actually
  release its socket, then relaunches it the same detached way
  EnsureRunning does.
- Fixed a real startup hang: resuming persisted rooms used to block
  before the daemon's Accept loop ever started, so a slow/unreachable
  relay in one persisted room could leave the daemon looking "alive" to
  IsRunning() while unable to answer any request — including its own
  successor's attempt to detect and clean up a stuck instance. Resuming
  now runs in the background instead. Added deadlines on both ends so a
  stuck daemon fails fast with an actionable message rather than hanging.
- Fixed peer counts being inflated by duplicate connections: two peers
  can end up with more than one link to each other (both sides happened
  to dial at once). Links now carry the remote peer's ID (known
  immediately when we dialed them, learned from their first manifest
  otherwise), and Status dedupes by that ID.
This commit is contained in:
2026-07-20 14:29:42 +08:00
parent c8f670e920
commit 84f3d62d68
9 changed files with 240 additions and 15 deletions
+1
View File
@@ -1,2 +1,3 @@
/reference/ /reference/
/test/ /test/
/share-*
BIN
View File
Binary file not shown.
+35
View File
@@ -7,6 +7,7 @@ import (
"path/filepath" "path/filepath"
"runtime" "runtime"
"strconv" "strconv"
"time"
"github.com/tzgyn/share/src/daemon" "github.com/tzgyn/share/src/daemon"
log "github.com/tzgyn/share/src/logger" log "github.com/tzgyn/share/src/logger"
@@ -40,11 +41,16 @@ USAGE EXAMPLES:
Stop the background daemon (leaves every room it's in): Stop the background daemon (leaves every room it's in):
share stop share stop
Restart the background daemon (e.g. after upgrading the binary) —
rooms it was in resume automatically:
share restart
COMMANDS: COMMANDS:
send seed file(s) into a room (see options with: share send -h) 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) relay start your own relay (optional; see options with: share relay -h)
status show the rooms the background daemon is currently in status show the rooms the background daemon is currently in
stop stop the background daemon and leave every room stop stop the background daemon and leave every room
restart stop and relaunch the background daemon
daemon run the background daemon in the foreground (mostly for debugging) daemon run the background daemon in the foreground (mostly for debugging)
Run "share [command] -h" for the options of a given command. Run "share [command] -h" for the options of a given command.
@@ -81,6 +87,8 @@ func Run() error {
return runRelay(splice(args, i)) return runRelay(splice(args, i))
case "stop": case "stop":
return runStop(splice(args, i)) return runStop(splice(args, i))
case "restart":
return runRestart(splice(args, i))
case "status": case "status":
return runStatus(splice(args, i)) return runStatus(splice(args, i))
case "daemon": case "daemon":
@@ -250,6 +258,33 @@ func runStop(args []string) error {
return nil return nil
} }
func runRestart(args []string) error {
fs := flag.NewFlagSet("restart", flag.ExitOnError)
fs.Usage = func() {
fmt.Fprintln(os.Stderr, "share restart [options]")
fs.PrintDefaults()
}
debug := fs.Bool("debug", false, "toggle debug mode for the relaunched daemon")
if err := fs.Parse(args); err != nil {
return err
}
if daemon.IsRunning() {
if err := daemon.Stop(); err != nil {
return fmt.Errorf("could not stop the running daemon: %w", err)
}
if err := daemon.WaitUntilStopped(5 * time.Second); err != nil {
return err
}
}
if err := daemon.EnsureRunning(*debug); err != nil {
return err
}
fmt.Println("Daemon restarted; any rooms it was in resume automatically.")
return nil
}
func runStatus(args []string) error { func runStatus(args []string) error {
fs := flag.NewFlagSet("status", flag.ExitOnError) fs := flag.NewFlagSet("status", flag.ExitOnError)
fs.Usage = func() { fs.Usage = func() {
+21 -1
View File
@@ -95,6 +95,21 @@ func Stop() error {
return err return err
} }
// WaitUntilStopped polls until the daemon is no longer reachable, or
// returns an error once timeout elapses. Useful right after Stop, since
// its response is sent before the daemon has necessarily finished tearing
// down and released its socket.
func WaitUntilStopped(timeout time.Duration) error {
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
if !IsRunning() {
return nil
}
time.Sleep(50 * time.Millisecond)
}
return fmt.Errorf("daemon did not stop in time")
}
// Status asks the running daemon which rooms it's currently in and how // Status asks the running daemon which rooms it's currently in and how
// each one's transfers are progressing. // each one's transfers are progressing.
func Status() ([]swarm.RoomStatus, error) { func Status() ([]swarm.RoomStatus, error) {
@@ -118,13 +133,18 @@ func call(req request) (response, error) {
return response{}, fmt.Errorf("could not reach daemon: %w", err) return response{}, fmt.Errorf("could not reach daemon: %w", err)
} }
defer conn.Close() defer conn.Close()
// A join can legitimately take a while (a real network dial to a
// relay); anything else should be near-instant. Either way, this must
// never hang forever — a daemon that's up but stuck for any reason
// should fail fast with a clear message instead.
conn.SetDeadline(time.Now().Add(35 * time.Second))
if err := json.NewEncoder(conn).Encode(req); err != nil { if err := json.NewEncoder(conn).Encode(req); err != nil {
return response{}, err return response{}, err
} }
var resp response var resp response
if err := json.NewDecoder(conn).Decode(&resp); err != nil { if err := json.NewDecoder(conn).Decode(&resp); err != nil {
return response{}, err return response{}, fmt.Errorf("daemon accepted the connection but didn't respond in time — it may be stuck; try `pkill -f \"share daemon\"` and, if that doesn't help, remove %s before relaunching: %w", path, err)
} }
if !resp.OK { if !resp.OK {
return resp, fmt.Errorf("%s", resp.Error) return resp, fmt.Errorf("%s", resp.Error)
+8
View File
@@ -21,3 +21,11 @@ func logPath() (string, error) {
} }
return filepath.Join(dir, "daemon.log"), nil return filepath.Join(dir, "daemon.log"), nil
} }
func statePath() (string, error) {
dir, err := utils.GetConfigDir(true)
if err != nil {
return "", err
}
return filepath.Join(dir, "rooms.json"), nil
}
+96
View File
@@ -0,0 +1,96 @@
package daemon
import (
"encoding/json"
"os"
"sync"
)
// stateMu guards rooms.json against concurrent read-modify-write from
// multiple join/completion calls.
var stateMu sync.Mutex
// loadPersistedRooms returns every room recorded as active from a
// previous run, so the daemon can rejoin them after a restart (crash,
// reboot, or an explicit "share stop" followed by relaunching). A
// missing file just means nothing to resume.
func loadPersistedRooms() ([]joinRequest, error) {
path, err := statePath()
if err != nil {
return nil, err
}
stateMu.Lock()
defer stateMu.Unlock()
return readRoomsLocked(path)
}
func readRoomsLocked(path string) ([]joinRequest, error) {
b, err := os.ReadFile(path)
if err != nil {
if os.IsNotExist(err) {
return nil, nil
}
return nil, err
}
var rooms []joinRequest
if err := json.Unmarshal(b, &rooms); err != nil {
// a corrupt state file shouldn't block the daemon from starting
return nil, nil
}
return rooms, nil
}
// savePersistedRoom records req as active, so it's resumed on the next
// restart. Joining the same code again just replaces the earlier entry.
func savePersistedRoom(req joinRequest) error {
return updatePersistedRooms(func(rooms []joinRequest) []joinRequest {
for i, r := range rooms {
if r.Code == req.Code {
rooms[i] = req
return rooms
}
}
return append(rooms, req)
})
}
// removePersistedRoom drops code from the resume list, so a future
// restart won't try to rejoin it.
func removePersistedRoom(code string) error {
return updatePersistedRooms(func(rooms []joinRequest) []joinRequest {
out := rooms[:0]
for _, r := range rooms {
if r.Code != code {
out = append(out, r)
}
}
return out
})
}
func updatePersistedRooms(mutate func([]joinRequest) []joinRequest) error {
path, err := statePath()
if err != nil {
return err
}
stateMu.Lock()
defer stateMu.Unlock()
rooms, err := readRoomsLocked(path)
if err != nil {
return err
}
rooms = mutate(rooms)
b, err := json.MarshalIndent(rooms, "", " ")
if err != nil {
return err
}
// write-then-rename so a crash mid-write can't corrupt the file that
// the next startup relies on
tmp := path + ".tmp"
if err := os.WriteFile(tmp, b, 0o600); err != nil {
return err
}
return os.Rename(tmp, path)
}
+45 -1
View File
@@ -6,6 +6,7 @@ import (
"net" "net"
"os" "os"
"sync" "sync"
"time"
log "github.com/tzgyn/share/src/logger" log "github.com/tzgyn/share/src/logger"
"github.com/tzgyn/share/src/swarm" "github.com/tzgyn/share/src/swarm"
@@ -43,6 +44,15 @@ func Serve() error {
s := &server{rooms: make(map[string]*swarm.Client)} s := &server{rooms: make(map[string]*swarm.Client)}
log.Infof("daemon listening on %s", path) log.Infof("daemon listening on %s", path)
// Resuming persisted rooms happens in the background, not before the
// Accept loop below: joining a room can block for a while (e.g. its
// relay is slow or unreachable), and this socket already looks
// "alive" to anyone calling IsRunning() the moment Listen succeeds
// above. If resuming blocked here, the daemon would appear up while
// unable to answer any request — including its own successor's
// attempt to detect and clean up a stuck instance.
go s.resumePersistedRooms()
for { for {
conn, err := l.Accept() conn, err := l.Accept()
if err != nil { if err != nil {
@@ -55,10 +65,30 @@ func Serve() error {
} }
} }
func (s *server) resumePersistedRooms() {
persisted, err := loadPersistedRooms()
if err != nil {
log.Warnf("could not load rooms to resume: %v", err)
return
}
for _, req := range persisted {
if err := s.join(req); err != nil {
log.Warnf("could not resume room %s: %v", req.Code, err)
continue
}
log.Infof("resumed room %s from a previous run", req.Code)
}
}
// handleConn services one control request and reports whether the // handleConn services one control request and reports whether the
// daemon should shut down after responding (true only for "stop"). // daemon should shut down after responding (true only for "stop").
func (s *server) handleConn(conn net.Conn) (shutdown bool) { func (s *server) handleConn(conn net.Conn) (shutdown bool) {
defer conn.Close() defer conn.Close()
// a control request is a single small read + write; nothing here
// should ever legitimately take long, so bound it defensively —
// join itself has no deadline (it can involve a real network dial),
// so give that specific case more room.
conn.SetDeadline(time.Now().Add(30 * time.Second))
var req request var req request
if err := json.NewDecoder(conn).Decode(&req); err != nil { if err := json.NewDecoder(conn).Decode(&req); err != nil {
@@ -120,6 +150,10 @@ func (s *server) join(req joinRequest) error {
s.mu.Unlock() s.mu.Unlock()
log.Infof("joined room %s", req.Code) log.Infof("joined room %s", req.Code)
if err := savePersistedRoom(req); err != nil {
log.Warnf("could not persist room %s (won't resume after a restart): %v", req.Code, err)
}
go func() { go func() {
err := cl.Wait() err := cl.Wait()
s.mu.Lock() s.mu.Lock()
@@ -127,8 +161,18 @@ func (s *server) join(req joinRequest) error {
s.mu.Unlock() s.mu.Unlock()
if err != nil { if err != nil {
log.Errorf("room %s ended: %v", req.Code, err) log.Errorf("room %s ended: %v", req.Code, err)
} else { } else if req.OneTime {
// finished on its own, not stopped — a one-shot fetch, don't
// resume it on the next restart
log.Infof("room %s finished", req.Code) log.Infof("room %s finished", req.Code)
if err := removePersistedRoom(req.Code); err != nil {
log.Warnf("could not remove persisted room %s: %v", req.Code, err)
}
} else {
// a non-one-time room only stops via an explicit Stop (i.e.
// "share stop", shutting the whole daemon down) — leave it
// persisted so it resumes next time the daemon starts.
log.Infof("room %s stopped", req.Code)
} }
}() }()
+28 -8
View File
@@ -56,7 +56,14 @@ type Client struct {
mu sync.Mutex mu sync.Mutex
files map[string]*sharedFile files map[string]*sharedFile
pending map[string]map[int64]*peerLink // fileID -> offset -> link we asked pending map[string]map[int64]*peerLink // fileID -> offset -> link we asked
links map[*peerLink]bool // links maps each open link to the remote peer's ID, once known. It's
// "" until the first message (always a manifest) arrives on that
// link — outbound links from connectToPeer know it immediately since
// we dialed that specific peer. Two links can end up pointing at the
// same peer (both sides happened to dial each other at once); Status
// dedupes by this ID so the reported peer count reflects distinct
// devices, not raw connections.
links map[*peerLink]string
dialedMu sync.Mutex dialedMu sync.Mutex
dialed map[string]bool // relay peer IDs already dialed, so we don't redial every peer-list update dialed map[string]bool // relay peer IDs already dialed, so we don't redial every peer-list update
@@ -76,7 +83,7 @@ func newClient(opts Options) *Client {
selfID: randomID(), selfID: randomID(),
files: make(map[string]*sharedFile), files: make(map[string]*sharedFile),
pending: make(map[string]map[int64]*peerLink), pending: make(map[string]map[int64]*peerLink),
links: make(map[*peerLink]bool), links: make(map[*peerLink]string),
dialed: make(map[string]bool), dialed: make(map[string]bool),
stopCh: make(chan struct{}), stopCh: make(chan struct{}),
done: make(chan error, 1), done: make(chan error, 1),
@@ -165,9 +172,15 @@ func (cl *Client) Status() RoomStatus {
Complete: f.store.Complete(), Complete: f.store.Complete(),
}) })
} }
distinctPeers := make(map[string]bool, len(cl.links))
for _, id := range cl.links {
if id != "" {
distinctPeers[id] = true
}
}
return RoomStatus{ return RoomStatus{
Code: cl.opts.Code, Code: cl.opts.Code,
Peers: len(cl.links), Peers: len(distinctPeers),
Files: files, Files: files,
} }
} }
@@ -279,7 +292,7 @@ func (cl *Client) acceptLoop(l net.Listener) {
conn.Close() conn.Close()
return return
} }
cl.handlePeerLink(newPeerLink(c, key)) cl.handlePeerLink(newPeerLink(c, key), "") // remote peer's ID isn't known until its manifest arrives
}(conn) }(conn)
} }
} }
@@ -342,7 +355,7 @@ func (cl *Client) connectToPeer(p tcp.PeerInfo) {
select { select {
case r := <-found: case r := <-found:
cl.handlePeerLink(newPeerLink(r.conn, r.key)) cl.handlePeerLink(newPeerLink(r.conn, r.key), p.ID) // we know exactly who we were trying to reach
case <-time.After(25 * time.Second): case <-time.After(25 * time.Second):
log.Warnf("could not connect to peer %s: direct dial and relay-piped fallback both failed", p.ID) log.Warnf("could not connect to peer %s: direct dial and relay-piped fallback both failed", p.ID)
} }
@@ -357,9 +370,9 @@ func pairKey(code, idA, idB string) string {
return code + ":" + idA + ":" + idB return code + ":" + idA + ":" + idB
} }
func (cl *Client) handlePeerLink(link *peerLink) { func (cl *Client) handlePeerLink(link *peerLink, knownID string) {
cl.mu.Lock() cl.mu.Lock()
cl.links[link] = true cl.links[link] = knownID
cl.mu.Unlock() cl.mu.Unlock()
log.Infof("connected to peer %s", link.remote) log.Infof("connected to peer %s", link.remote)
defer func() { defer func() {
@@ -404,12 +417,19 @@ func (cl *Client) sendManifest(link *peerLink) error {
files = append(files, e) files = append(files, e)
} }
cl.mu.Unlock() cl.mu.Unlock()
return sendMessage(link.conn, link.key, Message{Type: TypeManifest, Files: files}) return sendMessage(link.conn, link.key, Message{Type: TypeManifest, From: cl.selfID, Files: files})
} }
func (cl *Client) handleMessage(link *peerLink, msg Message) error { func (cl *Client) handleMessage(link *peerLink, msg Message) error {
switch msg.Type { switch msg.Type {
case TypeManifest: case TypeManifest:
if msg.From != "" {
cl.mu.Lock()
if cl.links[link] == "" {
cl.links[link] = msg.From
}
cl.mu.Unlock()
}
for _, entry := range msg.Files { for _, entry := range msg.Files {
cl.learnFile(entry) cl.learnFile(entry)
cl.considerRequests(entry.ID, entry.Have, link) cl.considerRequests(entry.ID, entry.Have, link)
+1
View File
@@ -42,6 +42,7 @@ type FileManifestEntry struct {
// Message is what flows over an established, encrypted peer link. // Message is what flows over an established, encrypted peer link.
type Message struct { type Message struct {
Type Type `json:"type"` Type Type `json:"type"`
From string `json:"from,omitempty"` // sender's peer ID; set on TypeManifest, the first message on every link
Files []FileManifestEntry `json:"files,omitempty"` // TypeManifest Files []FileManifestEntry `json:"files,omitempty"` // TypeManifest
FileID string `json:"fileId,omitempty"` FileID string `json:"fileId,omitempty"`
Offset int64 `json:"offset,omitempty"` Offset int64 `json:"offset,omitempty"`