diff --git a/.gitignore b/.gitignore index fb28d15..3f77e8b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ /reference/ /test/ +/share-* diff --git a/share b/share index 0d60016..b74453b 100755 Binary files a/share and b/share differ diff --git a/src/cli/cli.go b/src/cli/cli.go index 528804e..1ab43fe 100644 --- a/src/cli/cli.go +++ b/src/cli/cli.go @@ -7,6 +7,7 @@ import ( "path/filepath" "runtime" "strconv" + "time" "github.com/tzgyn/share/src/daemon" log "github.com/tzgyn/share/src/logger" @@ -40,12 +41,17 @@ USAGE EXAMPLES: Stop the background daemon (leaves every room it's in): share stop + Restart the background daemon (e.g. after upgrading the binary) — + rooms it was in resume automatically: + share restart + 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) + 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 + restart stop and relaunch the background daemon + daemon run the background daemon in the foreground (mostly for debugging) Run "share [command] -h" for the options of a given command. ` @@ -81,6 +87,8 @@ func Run() error { return runRelay(splice(args, i)) case "stop": return runStop(splice(args, i)) + case "restart": + return runRestart(splice(args, i)) case "status": return runStatus(splice(args, i)) case "daemon": @@ -250,6 +258,33 @@ func runStop(args []string) error { 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 { fs := flag.NewFlagSet("status", flag.ExitOnError) fs.Usage = func() { diff --git a/src/daemon/client.go b/src/daemon/client.go index 9a884d3..84df207 100644 --- a/src/daemon/client.go +++ b/src/daemon/client.go @@ -95,6 +95,21 @@ func Stop() error { 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 // each one's transfers are progressing. 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) } 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 { return response{}, err } var resp response 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 { return resp, fmt.Errorf("%s", resp.Error) diff --git a/src/daemon/paths.go b/src/daemon/paths.go index d8ecf18..3a71076 100644 --- a/src/daemon/paths.go +++ b/src/daemon/paths.go @@ -21,3 +21,11 @@ func logPath() (string, error) { } 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 +} diff --git a/src/daemon/persist.go b/src/daemon/persist.go new file mode 100644 index 0000000..08bbc62 --- /dev/null +++ b/src/daemon/persist.go @@ -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) +} diff --git a/src/daemon/server.go b/src/daemon/server.go index 943e888..536e167 100644 --- a/src/daemon/server.go +++ b/src/daemon/server.go @@ -6,6 +6,7 @@ import ( "net" "os" "sync" + "time" log "github.com/tzgyn/share/src/logger" "github.com/tzgyn/share/src/swarm" @@ -43,6 +44,15 @@ func Serve() error { s := &server{rooms: make(map[string]*swarm.Client)} 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 { conn, err := l.Accept() 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 // daemon should shut down after responding (true only for "stop"). func (s *server) handleConn(conn net.Conn) (shutdown bool) { 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 if err := json.NewDecoder(conn).Decode(&req); err != nil { @@ -120,6 +150,10 @@ func (s *server) join(req joinRequest) error { s.mu.Unlock() 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() { err := cl.Wait() s.mu.Lock() @@ -127,8 +161,18 @@ func (s *server) join(req joinRequest) error { s.mu.Unlock() if err != nil { 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) + 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) } }() diff --git a/src/swarm/client.go b/src/swarm/client.go index bac4904..9770232 100644 --- a/src/swarm/client.go +++ b/src/swarm/client.go @@ -56,7 +56,14 @@ type Client struct { mu sync.Mutex files map[string]*sharedFile 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 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(), files: make(map[string]*sharedFile), pending: make(map[string]map[int64]*peerLink), - links: make(map[*peerLink]bool), + links: make(map[*peerLink]string), dialed: make(map[string]bool), stopCh: make(chan struct{}), done: make(chan error, 1), @@ -165,9 +172,15 @@ func (cl *Client) Status() RoomStatus { Complete: f.store.Complete(), }) } + distinctPeers := make(map[string]bool, len(cl.links)) + for _, id := range cl.links { + if id != "" { + distinctPeers[id] = true + } + } return RoomStatus{ Code: cl.opts.Code, - Peers: len(cl.links), + Peers: len(distinctPeers), Files: files, } } @@ -279,7 +292,7 @@ func (cl *Client) acceptLoop(l net.Listener) { conn.Close() return } - cl.handlePeerLink(newPeerLink(c, key)) + cl.handlePeerLink(newPeerLink(c, key), "") // remote peer's ID isn't known until its manifest arrives }(conn) } } @@ -342,7 +355,7 @@ func (cl *Client) connectToPeer(p tcp.PeerInfo) { select { 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): 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 } -func (cl *Client) handlePeerLink(link *peerLink) { +func (cl *Client) handlePeerLink(link *peerLink, knownID string) { cl.mu.Lock() - cl.links[link] = true + cl.links[link] = knownID cl.mu.Unlock() log.Infof("connected to peer %s", link.remote) defer func() { @@ -404,12 +417,19 @@ func (cl *Client) sendManifest(link *peerLink) error { files = append(files, e) } 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 { switch msg.Type { 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 { cl.learnFile(entry) cl.considerRequests(entry.ID, entry.Have, link) diff --git a/src/swarm/message.go b/src/swarm/message.go index b8705b5..5264c07 100644 --- a/src/swarm/message.go +++ b/src/swarm/message.go @@ -42,6 +42,7 @@ type FileManifestEntry struct { // Message is what flows over an established, encrypted peer link. type Message struct { 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 FileID string `json:"fileId,omitempty"` Offset int64 `json:"offset,omitempty"`