// 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
own []string
notify func(Event)
connected bool
+ retrying bool
connectedAt time.Time
disconnectedAt time.Time
disconnections int
// 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
w.close()
}()
+ if linked {
+ connect()
+ }
+
for {
err := w.inbox.deliver(ctx, l, w.link(), false)
if err != nil {
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() {}
}
}
-// 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.
client := w.client
w.client = nil
w.linked = false
+ w.retrying = false
w.own = nil
w.connected = false
w.disconnectedAt = time.Time{}
// 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:
}
w.connected = true
+ w.retrying = false
w.connectedAt = w.clock()
if w.timer != nil {
w.timer.Stop()
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
}
// 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)
package main
import (
+ "context"
"errors"
"os"
"path/filepath"
"strings"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/require"
}
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) {
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) {
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,
+ }
+}
projects / onechat / commitdiff