From: Anderson A. Giacomolli Date: Mon, 9 Nov 2026 17:48:10 +0000 (-0300) Subject: Retry the first WhatsApp connection X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/53c6c8a5a17e2dfc8bd3ff02ab99d29d709dda2f Retry the first WhatsApp connection Keep failed starting connections in the Run select with a timer. Start at 2 seconds and double the wait up to 5 minutes, so the inbox and commands stay active during waits. End retries on a connection, context end, or link change. Share outage times, notices, status, and redacted logs with lost connections. Keep one failure line and one recovery line for an outage. Disable library startup retries only for linked devices. A returned ConnectContext error has no Disconnected event, while the library retry mode emits that event and returns nil. One retry owner prevents parallel loops and gives the requested delay cap. Pairing and retries after an open connection keep their library behavior. --- diff --git a/README.rst b/README.rst index 9b9c489..0cc4b1e 100644 --- a/README.rst +++ b/README.rst @@ -357,12 +357,16 @@ Each batch of a group goes to the chat, also with no turn, and the memory tree summarizes it with model calls, which use the quota. When the session holds a linked device, the daemon connects at its -start, and after a network failure it connects again by itself. When a -disconnection lasts 10 minutes, all clients get one notice, and one more -when the connection comes back. When WhatsApp ends the connection for -good, for example because another program uses the same session, all -clients get a notice at once. When the user removes the device in the -phone, the daemon deletes the session and tells all clients one time. +start. If the first connection fails, it tries again after 2 seconds, +then doubles the wait up to 5 minutes. The inbox and commands stay +active during each wait. An unlink, logout, or pair command stops these +retries. After a network failure of an open connection, the library +connects again by itself. When a disconnection lasts 10 minutes, all +clients get one notice, and one more when the connection comes back. +When WhatsApp ends the connection for good, for example because another +program uses the same session, all clients get a notice at once. When +the user removes the device in the phone, the daemon deletes the session +and tells all clients one time. *********** Schedules diff --git a/whatsapp.go b/whatsapp.go index 40729d8..e02e8c6 100644 --- a/whatsapp.go +++ b/whatsapp.go @@ -65,6 +65,13 @@ const ( // gets a notice. whatsappNoticeDelay = 10 * time.Minute + // whatsappRetryDelay is the first wait after a failed starting + // connection. + whatsappRetryDelay = 2 * time.Second + + // whatsappRetryLimit is the longest wait between starting connections. + whatsappRetryLimit = 5 * time.Minute + // whatsappRequests is the number of pair and unlink commands that can // wait for Run. whatsappRequests = 8 @@ -187,6 +194,7 @@ type WhatsApp struct { own []string notify func(Event) connected bool + retrying bool connectedAt time.Time disconnectedAt time.Time disconnections int @@ -198,18 +206,41 @@ type WhatsApp struct { // Run connects a linked device, logs the closed batches in l, runs the pair and // unlink commands, and handles a logout from the phone, until ctx ends. Its // notes go to the log of l, and notify gets the notices for all clients. A -// failed first connection is logged, and the library tries again when the -// failure is one of the network. Run returns the error of a write of the log or -// of the state file, which stops the daemon, because then messages would be -// lost. +// failed first connection starts the same outage notice as a lost connection. A +// timer in the select retries it with a doubled wait from 2 seconds to 5 +// minutes, without a retry goroutine or a wait that blocks commands or the +// inbox. A connection, unlink, logout, pair command, or ctx end stops retries. +// The library owns retries after a connection opens. Run returns the error of a +// write of the log or of the state file, which stops the daemon, because then +// messages would be lost. func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) error { w.mutex.Lock() w.notify = notify linked := w.linked + w.retrying = linked w.mutex.Unlock() - if linked { - w.connect(ctx) + timer := time.NewTimer(whatsappRetryDelay) + timer.Stop() + defer timer.Stop() + + var retry <-chan time.Time + delay := whatsappRetryDelay + connect := func() { + retry = nil + if w.connect(ctx) { + timer.Reset(delay) + retry = timer.C + delay = min(2*delay, whatsappRetryLimit) + } + } + + stopRetry := func() { + timer.Stop() + retry = nil + w.mutex.Lock() + w.retrying = false + w.mutex.Unlock() } var pairing sync.WaitGroup @@ -220,6 +251,10 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err w.close() }() + if linked { + connect() + } + for { err := w.inbox.deliver(ctx, l, w.link(), false) if err != nil { @@ -233,12 +268,16 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err return err case <-w.inbox.wake: case <-w.inbox.due(): + case <-retry: + connect() case <-w.logouts: + stopRetry() err = w.loggedOut(ctx, l) if err != nil { return err } case request := <-w.requests: + stopRetry() cancel() pairing.Wait() cancel = func() {} @@ -396,17 +435,59 @@ func (w *WhatsApp) link() WhatsAppLink { } } -// connect makes the client of the linked device and connects it. -func (w *WhatsApp) connect(ctx context.Context) { +// connect tries the starting connection of the linked device with one client. +// It returns whether Run must retry. A failed call starts the shared outage +// state and logs its redacted cause once, because a returned ConnectContext +// error does not emit Disconnected. Connected logs recovery after +// authentication. +func (w *WhatsApp) connect(ctx context.Context) bool { w.mutex.Lock() - w.client = w.newClient(w.device, w.handle) + if ctx.Err() != nil || !w.linked || !w.retrying || w.closed { + w.mutex.Unlock() + return false + } + + if w.client == nil { + w.client = w.newClient(w.device, w.handle) + } + client := w.client w.mutex.Unlock() err := client.ConnectContext(ctx) - if err != nil { - w.logger.Printf("connect failed, restart to retry: %s", w.redact(err.Error())) + w.mutex.Lock() + if err == nil || ctx.Err() != nil || !w.retrying || !w.linked { + w.retrying = false + w.mutex.Unlock() + return false } + + first := w.startOutage() + cause := whatsappRedact(w.own, err.Error()) + w.mutex.Unlock() + if first { + w.logger.Printf("first connect failed, retrying: %s", cause) + } + + return true +} + +// startOutage starts the shared outage time and notice timer once. The caller +// holds the mutex. Its result tells whether this is the first failure. +func (w *WhatsApp) startOutage() bool { + first := w.linked && (w.connected || w.disconnectedAt.IsZero()) + if first { + w.disconnectedAt = w.clock() + w.disconnections++ + disconnections := w.disconnections + w.timer = time.AfterFunc(w.noticeDelay, func() { + w.disconnectedLong(disconnections) + }) + } + + w.connected = false + + return first } // close ends the connection and the timer of the notice, and stops all notices. @@ -514,6 +595,7 @@ func (w *WhatsApp) removeSession() error { client := w.client w.client = nil w.linked = false + w.retrying = false w.own = nil w.connected = false w.disconnectedAt = time.Time{} @@ -695,14 +777,14 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) { // goes to the inbox. The names of the groups follow the events of the groups. A // connection starts a catch-up of the inbox, which ends at OfflineSyncCompleted // or after its catchupLimit. A disconnection keeps the time of its start while -// the library tries again. After noticeDelay of a disconnection, all clients -// get one notice, and they get one more when the connection comes back. A -// disconnection that the library does not try again gives a notice at once. +// Run or the library tries again. After noticeDelay of a disconnection, all +// clients get one notice, and they get one more when the connection comes back. +// A disconnection that the library does not try again gives a notice at once. // // The log of the daemon gets one line for each change of the connection of the // linked device: connected, a failed first connection, a lost connection, and -// connected again with the length of the outage. The retries of the library -// between them log nothing. +// connected again with the length of the outage. The retries of the library and +// Run between them log nothing. func (w *WhatsApp) handle(event any) bool { switch event := event.(type) { case *events.Message: @@ -738,6 +820,7 @@ func (w *WhatsApp) handle(event any) bool { } w.connected = true + w.retrying = false w.connectedAt = w.clock() if w.timer != nil { w.timer.Stop() @@ -748,27 +831,21 @@ func (w *WhatsApp) handle(event any) bool { notice = "WhatsApp is connected again." } case *events.Disconnected: - if w.linked && (w.connected || w.disconnectedAt.IsZero()) { + connected := w.connected + if w.startOutage() { line = "disconnected, retrying" - if !w.connected { + if !connected { line = "first connect failed, retrying" } - - w.disconnectedAt = w.clock() - w.disconnections++ - disconnections := w.disconnections - w.timer = time.AfterFunc(w.noticeDelay, func() { - w.disconnectedLong(disconnections) - }) } - - w.connected = false case *events.LoggedOut: + w.retrying = false select { case w.logouts <- struct{}{}: default: } case events.PermanentDisconnect: + w.retrying = false if w.linked { w.connected = false w.noticed = true @@ -862,12 +939,13 @@ func whatsappSetDeviceProps() { } // whatsappNewClient returns the client of the library for device, with handle -// as its event handler. The client tries again after a failed first connection, -// acknowledges a message only after handle returns true, and downloads no old -// history. +// as its event handler. Run owns failed starting connections of linked devices, +// so only an unlinked device enables the initial retry of the library for +// pairing. The library still retries a lost connection. The client acknowledges +// a message only after handle returns true, and downloads no old history. func whatsappNewClient(device *store.Device, handle func(event any) bool) WhatsAppClient { client := whatsmeow.NewClient(device, waLog.Noop) - client.InitialAutoReconnect = true + client.InitialAutoReconnect = device.ID == nil client.SynchronousAck = true client.ManualHistorySyncDownload = true client.AddEventHandlerWithSuccessStatus(handle) diff --git a/whatsapp_fake_helper_test.go b/whatsapp_fake_helper_test.go index ee5321e..7aa558b 100644 --- a/whatsapp_fake_helper_test.go +++ b/whatsapp_fake_helper_test.go @@ -69,29 +69,32 @@ var ( // daemon gives it, and Emit gives events to the handler, as the library does. // Each connection keeps its context and sends to connects, and each pairing // sends its channel to pairings, so that the test gives the items of the -// pairing. A connection fails with connectErr, a logout with logoutErr, and a -// send with sendErr. groups are the groups of the account. A send that does not -// fail gets the time whatsappTestSendTime. +// pairing. A connection takes its error from connectErrors, then connectErr. A +// logout fails with logoutErr, and a send with sendErr. groups are the groups +// of the account. A send that does not fail gets the time whatsappTestSendTime. type WhatsAppFake struct { connects chan struct{} pairings chan chan whatsmeow.QRChannelItem - mutex sync.Mutex - handle func(event any) bool - connectErr error - logoutErr error - groups []*types.GroupInfo - sendErr error - phones []string - logouts int - sent []WhatsAppFakeSend - connections []context.Context + mutex sync.Mutex + handle func(event any) bool + connectErr error + connectErrors []error + connectTimes []time.Time + logoutErr error + groups []*types.GroupInfo + sendErr error + phones []string + logouts int + sent []WhatsAppFakeSend + connections []context.Context } // ConnectContext keeps ctx and tells the test that the daemon connected. func (f *WhatsAppFake) ConnectContext(ctx context.Context) error { f.mutex.Lock() f.connections = append(f.connections, ctx) + f.connectTimes = append(f.connectTimes, time.Now()) f.mutex.Unlock() f.connects <- struct{}{} @@ -99,6 +102,12 @@ func (f *WhatsAppFake) ConnectContext(ctx context.Context) error { f.mutex.Lock() defer f.mutex.Unlock() + if len(f.connectErrors) > 0 { + err := f.connectErrors[0] + f.connectErrors = f.connectErrors[1:] + return err + } + return f.connectErr } @@ -110,6 +119,14 @@ func (f *WhatsAppFake) Connections() []context.Context { return slices.Clone(f.connections) } +// ConnectionTimes returns the time of each connection call, in order. +func (f *WhatsAppFake) ConnectionTimes() []time.Time { + f.mutex.Lock() + defer f.mutex.Unlock() + + return slices.Clone(f.connectTimes) +} + // Disconnect does nothing. func (f *WhatsAppFake) Disconnect() { } diff --git a/whatsapp_test.go b/whatsapp_test.go index 7001ea1..c14571b 100644 --- a/whatsapp_test.go +++ b/whatsapp_test.go @@ -1,11 +1,13 @@ package main import ( + "context" "errors" "os" "path/filepath" "strings" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -35,15 +37,210 @@ func TestWhatsAppStatusFollowsTheConnection(t *testing.T) { } func TestWhatsAppStatusOfAFirstConnectionThatFailed(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.fake.connectErr = errors.New("network unavailable") + test.start(t) + synctest.Wait() + + require.Equal(t, "WhatsApp is linked and disconnected since "+time.Now().Format(whatsappTimeLayout)+".", test.whatsapp.Status()) + require.Equal(t, "first connect failed, retrying: network unavailable\n", test.logs.String()) + }) +} + +func TestWhatsAppRetriesTheFirstConnectionWithCappedBackoff(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + for range 10 { + test.fake.connectErrors = append(test.fake.connectErrors, errors.New("network unavailable")) + } + + test.start(t) + synctest.Wait() + require.Len(t, test.fake.Connections(), 1) + require.True(t, test.fake.Emit(&events.Disconnected{})) + require.True(t, test.fake.Emit(&events.Disconnected{})) + + delays := []time.Duration{2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, 32 * time.Second, 64 * time.Second, 128 * time.Second, 256 * time.Second, 5 * time.Minute, 5 * time.Minute} + var outage time.Duration + for index, delay := range delays { + time.Sleep(delay - time.Nanosecond) + synctest.Wait() + require.Len(t, test.fake.Connections(), index+1) + time.Sleep(time.Nanosecond) + synctest.Wait() + require.Len(t, test.fake.Connections(), index+2) + outage += delay + } + + times := test.fake.ConnectionTimes() + for index, delay := range delays { + require.Equal(t, delay, times[index+1].Sub(times[index])) + } + + require.True(t, test.fake.Emit(&events.Connected{})) + require.True(t, test.fake.Emit(&events.OfflineSyncCompleted{})) + synctest.Wait() + time.Sleep(10 * time.Minute) + synctest.Wait() + require.Len(t, test.fake.Connections(), 11) + require.Equal(t, "first connect failed, retrying: network unavailable\nconnected again after "+durationText(outage)+"\n", test.logs.String()) + }) +} + +func TestWhatsAppFirstConnectionRetriesStopWhenContextEnds(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.fake.connectErr = errors.New("network unavailable") + test.start(t) + synctest.Wait() + time.Sleep(2 * time.Second) + synctest.Wait() + require.Len(t, test.fake.Connections(), 2) + + test.cancel() + require.NoError(t, <-test.done) + time.Sleep(10 * time.Minute) + synctest.Wait() + require.Len(t, test.fake.Connections(), 2) + }) +} + +func TestWhatsAppFirstConnectionRetriesStopForLinkChanges(t *testing.T) { + testCases := []struct { + name string + stop func(test *WhatsAppRetryTest) + }{ + { + name: "unlink", + stop: func(test *WhatsAppRetryTest) { + test.whatsapp.Command("unlink", func(text string) {}) + }, + }, + { + name: "logout", + stop: func(test *WhatsAppRetryTest) { + test.fake.Emit(&events.LoggedOut{}) + }, + }, + { + name: "pair command", + stop: func(test *WhatsAppRetryTest) { + test.whatsapp.Command("pair 15550001111", func(text string) {}) + }, + }, + { + name: "permanent disconnect", + stop: func(test *WhatsAppRetryTest) { + test.fake.Emit(&events.StreamReplaced{}) + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.fake.connectErr = errors.New("network unavailable") + test.start(t) + synctest.Wait() + tc.stop(test) + synctest.Wait() + time.Sleep(10 * time.Minute) + synctest.Wait() + require.Len(t, test.fake.Connections(), 1) + }) + }) + } +} + +func TestWhatsAppPairsAfterFirstConnectionRetriesStop(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.fake.connectErr = errors.New("network unavailable") + test.start(t) + synctest.Wait() + test.whatsapp.Command("unlink", func(text string) {}) + synctest.Wait() + require.False(t, test.whatsapp.isLinked()) + + test.fake.mutex.Lock() + test.fake.connectErr = nil + test.fake.mutex.Unlock() + test.whatsapp.Command("pair 15550001111", func(text string) {}) + items := <-test.fake.pairings + items <- whatsmeow.QRChannelItem{Event: whatsmeow.QRChannelEventCode} + synctest.Wait() + require.True(t, test.fake.Emit(&events.PairSuccess{ + ID: types.NewADJID(whatsappTestNumber, 0, 1), + LID: types.NewADJID(whatsappTestLID, 0, 1), + })) + items <- whatsmeow.QRChannelSuccess + synctest.Wait() + require.True(t, test.whatsapp.isLinked()) + time.Sleep(10 * time.Minute) + synctest.Wait() + require.Len(t, test.fake.Connections(), 2) + require.Equal(t, []string{whatsappTestNumber}, test.fake.Phones()) + }) +} + +func TestWhatsAppInboxFailureStopsFirstConnectionRetries(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.fake.connectErr = errors.New("network unavailable") + test.start(t) + synctest.Wait() + failure := errors.New("state write failed") + test.whatsapp.inbox.failures <- failure + require.ErrorIs(t, <-test.done, failure) + time.Sleep(10 * time.Minute) + synctest.Wait() + require.Len(t, test.fake.Connections(), 1) + }) +} + +func TestWhatsAppLongFirstConnectionOutageGivesOneNotice(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.fake.connectErr = errors.New("network unavailable") + test.start(t) + synctest.Wait() + since := time.Now().Format(whatsappTimeLayout) + time.Sleep(whatsappNoticeDelay) + synctest.Wait() + require.Equal(t, 1, test.events.Count("notice")) + require.Equal(t, "WhatsApp is disconnected since "+since+". OneChat tries again.", test.events.Last("notice")) + + time.Sleep(whatsappNoticeDelay) + synctest.Wait() + require.Equal(t, 1, test.events.Count("notice")) + test.fake.mutex.Lock() + test.fake.connectErr = nil + test.fake.mutex.Unlock() + time.Sleep(whatsappRetryLimit) + synctest.Wait() + require.True(t, test.fake.Emit(&events.Connected{})) + require.True(t, test.fake.Emit(&events.OfflineSyncCompleted{})) + synctest.Wait() + require.Equal(t, 2, test.events.Count("notice")) + require.Equal(t, "WhatsApp is connected again.", test.events.Last("notice")) + }) +} + +func TestWhatsAppStartingRetriesDoNotDuplicateLibraryRetries(t *testing.T) { dir := t.TempDir() whatsappTestSession(t, dir, true) - whatsapp, fake, logs := whatsappStartTest(t, dir, newTestClock()) - daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp) - - <-fake.connects - require.True(t, fake.Emit(&events.Disconnected{})) - require.Equal(t, "WhatsApp is linked and disconnected since 2026-01-01 12:00.", whatsapp.Status()) - require.Equal(t, "first connect failed, retrying\n", logs.String()) + whatsapp, _, _ := whatsappStartTest(t, dir, newTestClock()) + client, ok := whatsappNewClient(whatsapp.device, whatsapp.handle).(*whatsmeow.Client) + require.True(t, ok) + require.False(t, client.InitialAutoReconnect) + require.True(t, client.EnableAutoReconnect) + + device := whatsapp.container.NewDevice() + client, ok = whatsappNewClient(device, whatsapp.handle).(*whatsmeow.Client) + require.True(t, ok) + require.True(t, client.InitialAutoReconnect) } func TestWhatsAppLogsEachChangeOfTheConnectionOnce(t *testing.T) { @@ -396,7 +593,7 @@ func TestWhatsAppLogsNoOwnNumber(t *testing.T) { require.Eventually(t, func() bool { return logs.String() != "" }, 10*time.Second, time.Millisecond) - require.Equal(t, "connect failed, restart to retry: no route for OWN@s.whatsapp.net and OWN@lid\n", logs.String()) + require.Equal(t, "first connect failed, retrying: no route for OWN@s.whatsapp.net and OWN@lid\n", logs.String()) } func TestWhatsAppPhone(t *testing.T) { @@ -487,3 +684,60 @@ func TestSubagentHasNoWhatsAppTools(t *testing.T) { require.Equal(t, `error: unknown tool "whatsapp_find"`, box.run(t.Context(), "whatsapp_find", `{"name":"ana"}`)) require.Equal(t, `error: unknown tool "whatsapp_send"`, box.run(t.Context(), "whatsapp_send", `{"id":"a","chat":"15550002222@s.whatsapp.net","text":"hi"}`)) } + +// WhatsAppRetryTest runs WhatsApp without a model or network client call. All +// timers and channels belong to its synctest bubble, so backoff uses fake time. +type WhatsAppRetryTest struct { + whatsapp *WhatsApp + fake *WhatsAppFake + logs *SyncBuffer + loop *TurnLoop + events *TurnEventRecorder + cancel context.CancelFunc + done chan error +} + +func (r *WhatsAppRetryTest) start(t *testing.T) { + t.Helper() + + ctx, cancel := context.WithCancel(t.Context()) + r.cancel = cancel + r.done = make(chan error, 1) + go func() { + r.done <- r.whatsapp.Run(ctx, r.loop, r.events.Add) + }() + + t.Cleanup(func() { + cancel() + synctest.Wait() + }) +} + +func whatsappRetryTest(t *testing.T) *WhatsAppRetryTest { + t.Helper() + + dir := t.TempDir() + whatsappTestSession(t, dir, true) + whatsapp, fake, logs := whatsappStartTest(t, dir, newTestClock()) + whatsapp.clock = time.Now + agent, err := openAgent(t.TempDir()) + require.NoError(t, err) + memory, err := openMemory(t.Context(), agent.dir, time.Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, memory.Close()) + }) + + events := &TurnEventRecorder{} + client := newModelClient("https://example.com", ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) + loop, err := newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, time.Now), client, time.Now, discardLog(), events.Add) + require.NoError(t, err) + + return &WhatsAppRetryTest{ + whatsapp: whatsapp, + fake: fake, + logs: logs, + loop: loop, + events: events, + } +}