ipc: promote startup socket-wait to a shared DialWait; use in all modules
The cold-start crash-loop wasn't mavweb-specific — mavpoll and mavcaldav also ipc.Dial + exit on failure, so they crash-looped until core booted too. Moved the retry into ipc.DialWait (capped backoff, bounded) and switched mavweb, mavpoll, mavcaldav to it. mavweb's local dialCoreWithRetry is gone. Test: server appears after DialWait starts → it waits and connects. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -58,7 +58,7 @@ func run(args []string) error {
|
|||||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
core, err := ipc.Dial(*socket)
|
core, err := ipc.DialWait(*socket, 60*time.Second)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -69,7 +69,7 @@ func run(args []string) error {
|
|||||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
core, err := ipc.Dial(*socket)
|
core, err := ipc.DialWait(*socket, 60*time.Second)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-31
@@ -87,7 +87,7 @@ func main() {
|
|||||||
|
|
||||||
var core ipc.CoreAPI
|
var core ipc.CoreAPI
|
||||||
if *coreSock != "" {
|
if *coreSock != "" {
|
||||||
c, err := dialCoreWithRetry(*coreSock, 60*time.Second)
|
c, err := ipc.DialWait(*coreSock, 60*time.Second)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("dial core %s: %v", *coreSock, err)
|
log.Fatalf("dial core %s: %v", *coreSock, err)
|
||||||
}
|
}
|
||||||
@@ -169,36 +169,6 @@ func main() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// dialCoreWithRetry waits for core's IPC socket to appear before giving up.
|
|
||||||
// Core loads models on boot and (esp. under compose) may come up after mavweb;
|
|
||||||
// depends_on only orders container start, not socket readiness. Retrying with
|
|
||||||
// capped backoff up to timeout stops mavweb crash-looping on a cold start. A
|
|
||||||
// core still absent past the deadline is genuinely fatal. Mid-life core
|
|
||||||
// restarts are handled separately by ipc.Client's redial-on-drop.
|
|
||||||
func dialCoreWithRetry(path string, timeout time.Duration) (*ipc.Client, error) {
|
|
||||||
deadline := time.Now().Add(timeout)
|
|
||||||
delay := 200 * time.Millisecond
|
|
||||||
for attempt := 1; ; attempt++ {
|
|
||||||
c, err := ipc.Dial(path)
|
|
||||||
if err == nil {
|
|
||||||
if attempt > 1 {
|
|
||||||
log.Printf("core socket %s ready after %d attempts", path, attempt)
|
|
||||||
}
|
|
||||||
return c, nil
|
|
||||||
}
|
|
||||||
if time.Now().After(deadline) {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
if attempt == 1 {
|
|
||||||
log.Printf("waiting for core socket %s ...", path)
|
|
||||||
}
|
|
||||||
time.Sleep(delay)
|
|
||||||
if delay < 2*time.Second {
|
|
||||||
delay *= 2
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func handleWS(w http.ResponseWriter, r *http.Request, voiceAddr string) {
|
func handleWS(w http.ResponseWriter, r *http.Request, voiceAddr string) {
|
||||||
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
|
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
|
||||||
OriginPatterns: []string{"*"},
|
OriginPatterns: []string{"*"},
|
||||||
|
|||||||
@@ -55,6 +55,30 @@ func (c *Client) Close() error {
|
|||||||
return c.conn.Close()
|
return c.conn.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DialWait is Dial with patience: it retries with capped backoff until the
|
||||||
|
// socket is reachable or timeout elapses. Core loads models on boot and may
|
||||||
|
// come up after its modules (compose depends_on orders container start, not
|
||||||
|
// socket readiness), so a module that Dial'd once would crash-loop on a cold
|
||||||
|
// start. Every core-dialing module should use this instead of Dial. Mid-life
|
||||||
|
// core restarts are handled separately by the Client's own redial-on-drop.
|
||||||
|
func DialWait(path string, timeout time.Duration) (*Client, error) {
|
||||||
|
deadline := time.Now().Add(timeout)
|
||||||
|
delay := 200 * time.Millisecond
|
||||||
|
for {
|
||||||
|
c, err := Dial(path)
|
||||||
|
if err == nil {
|
||||||
|
return c, nil
|
||||||
|
}
|
||||||
|
if time.Now().After(deadline) {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
time.Sleep(delay)
|
||||||
|
if delay < 2*time.Second {
|
||||||
|
delay *= 2
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// call — the single request/response engine. Serialized by c.mu so a frame
|
// call — the single request/response engine. Serialized by c.mu so a frame
|
||||||
// and its reply always pair up; no interleaving to disambiguate. A wire
|
// and its reply always pair up; no interleaving to disambiguate. A wire
|
||||||
// RpcError is rehydrated into the matching package sentinel (errors.Is works
|
// RpcError is rehydrated into the matching package sentinel (errors.Is works
|
||||||
|
|||||||
@@ -57,6 +57,46 @@ func newServerWithStore(t *testing.T) (CoreAPI, *Server, *Client, *store.Store)
|
|||||||
return api, srv, cli, s
|
return api, srv, cli, s
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestDialWait_WaitsForLateServer — a module may start before core's socket
|
||||||
|
// exists (core loads models first). DialWait must keep retrying until the
|
||||||
|
// socket appears rather than fail on the first attempt.
|
||||||
|
func TestDialWait_WaitsForLateServer(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
sock := filepath.Join(dir, "maven.sock")
|
||||||
|
srvCh := make(chan *Server, 1)
|
||||||
|
errCh := make(chan error, 1)
|
||||||
|
|
||||||
|
// bring the server up only after DialWait is already retrying
|
||||||
|
go func() {
|
||||||
|
time.Sleep(300 * time.Millisecond)
|
||||||
|
s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
|
||||||
|
if err != nil {
|
||||||
|
errCh <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = s.Close() })
|
||||||
|
srv, err := Listen(sock, NewStoreAPI(s))
|
||||||
|
if err != nil {
|
||||||
|
errCh <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
go func() { _ = srv.Serve() }()
|
||||||
|
srvCh <- srv
|
||||||
|
}()
|
||||||
|
|
||||||
|
cli, err := DialWait(sock, 5*time.Second)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DialWait should connect once the server appears: %v", err)
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case err := <-errCh:
|
||||||
|
t.Fatalf("server setup failed: %v", err)
|
||||||
|
case srv := <-srvCh:
|
||||||
|
_ = cli.Close() // close client first so srv.Close's handler wait returns
|
||||||
|
_ = srv.Close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestClient_ReconnectsAfterServerRestart — a long-lived module (e.g. mavweb)
|
// TestClient_ReconnectsAfterServerRestart — a long-lived module (e.g. mavweb)
|
||||||
// must survive a core restart. The first server is closed and a new one is
|
// must survive a core restart. The first server is closed and a new one is
|
||||||
// brought up on the SAME socket path (as a daemon restart does); the client's
|
// brought up on the SAME socket path (as a daemon restart does); the client's
|
||||||
|
|||||||
Reference in New Issue
Block a user