From 483ca99c4609adfb64a905a7054f10945d68e761 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 28 Jul 2026 21:11:25 -0300 Subject: [PATCH 1/2] fix: harden instance lifecycle and QR retries --- pkg/instance/service/instance_service.go | 4 +- pkg/whatsmeow/service/lifecycle_test.go | 61 ++++ pkg/whatsmeow/service/whatsmeow.go | 428 ++++++++++++++--------- 3 files changed, 323 insertions(+), 170 deletions(-) create mode 100644 pkg/whatsmeow/service/lifecycle_test.go diff --git a/pkg/instance/service/instance_service.go b/pkg/instance/service/instance_service.go index fa43c946..6d40ce70 100644 --- a/pkg/instance/service/instance_service.go +++ b/pkg/instance/service/instance_service.go @@ -258,7 +258,7 @@ func (i instances) Connect(data *ConnectStruct, instance *instance_model.Instanc if !isInstanceRunning { i.loggerWrapper.GetLogger(instance.Id).LogInfo("[%s] Starting new client instance", instance.Id) - i.killChannel[instance.Id] = make(chan bool) + i.killChannel[instance.Id] = make(chan bool, 1) clientData := &whatsmeow_service.ClientData{ Instance: instance, @@ -713,7 +713,7 @@ func (i instances) ForceReconnect(instanceId string, number string) error { subscribedEvents := strings.Split(instance.Events, ",") - i.killChannel[instance.Id] = make(chan bool) + i.killChannel[instance.Id] = make(chan bool, 1) clientData := &whatsmeow_service.ClientData{ Instance: instance, diff --git a/pkg/whatsmeow/service/lifecycle_test.go b/pkg/whatsmeow/service/lifecycle_test.go new file mode 100644 index 00000000..1c51f0d2 --- /dev/null +++ b/pkg/whatsmeow/service/lifecycle_test.go @@ -0,0 +1,61 @@ +package whatsmeow_service + +import ( + "fmt" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestCanAutoReconnectRequiresPairedJID(t *testing.T) { + for _, jid := range []string{"", " ", " "} { + if canAutoReconnect(jid) { + t.Fatalf("unpaired jid %q must not auto-reconnect", jid) + } + } + if !canAutoReconnect("paired-device@s.whatsapp.net") { + t.Fatal("paired jid must auto-reconnect") + } +} + +func TestReserveReconnectDeduplicatesConcurrentAttempts(t *testing.T) { + service := &whatsmeowService{reconnectState: make(map[string]*instanceReconnectState)} + var accepted atomic.Int32 + var wg sync.WaitGroup + for attempt := 0; attempt < 100; attempt++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, ok := service.reserveReconnect("tenant-2"); ok { + accepted.Add(1) + } + }() + } + wg.Wait() + if got := accepted.Load(); got != 1 { + t.Fatalf("expected exactly one concurrent reconnect, got %d", got) + } + + service.finishReconnect("tenant-2", false) + if _, ok := service.reserveReconnect("tenant-2"); !ok { + t.Fatal("a new retry must be accepted after the prior attempt finishes") + } +} + +func TestReconnectDelayIsBoundedAndJitteredAcrossTwentyTenants(t *testing.T) { + seen := map[time.Duration]struct{}{} + for tenant := 1; tenant <= 20; tenant++ { + delay := reconnectDelay(fmt.Sprintf("tenant-%d", tenant), 0) + if delay < 2*time.Second || delay >= 2400*time.Millisecond { + t.Fatalf("tenant %d first delay out of bounds: %s", tenant, delay) + } + seen[delay] = struct{}{} + } + if len(seen) < 5 { + t.Fatalf("expected retry jitter across tenants, got only %d distinct delays", len(seen)) + } + if delay := reconnectDelay("tenant-20", 1000); delay < 5*time.Minute || delay >= 6*time.Minute { + t.Fatalf("capped retry delay out of bounds: %s", delay) + } +} diff --git a/pkg/whatsmeow/service/whatsmeow.go b/pkg/whatsmeow/service/whatsmeow.go index 366f0edb..22b68b24 100644 --- a/pkg/whatsmeow/service/whatsmeow.go +++ b/pkg/whatsmeow/service/whatsmeow.go @@ -7,6 +7,7 @@ import ( "encoding/base64" "encoding/json" "fmt" + "hash/fnv" "image/png" "io" "math/rand" @@ -55,6 +56,8 @@ type WhatsmeowService interface { ConnectOnStartup(clientName string) StartInstance(instanceId string) error ReconnectClient(instanceId string) error + ScheduleReconnect(instanceId string) + markReconnectSuccess(instanceId string) ClearInstanceCache(instanceId string, token string) error CallWebhook(instance *instance_model.Instance, queueName string, jsonData []byte) SendToGlobalQueues(event string, jsonData []byte, userId string) @@ -97,6 +100,74 @@ type whatsmeowService struct { natsProducer producer_interfaces.Producer loggerWrapper *logger_wrapper.LoggerManager passkeyCeremony *ceremony.Store + runtimeMu sync.Mutex + doneChannel map[string]chan struct{} + reconnectMu sync.Mutex + reconnectState map[string]*instanceReconnectState + authContainerOnce sync.Once + authContainer *sqlstore.Container + authContainerErr error +} + +type instanceReconnectState struct { + scheduled bool + failures int +} + +const reconnectProbeTimeout = 15 * time.Second + +func reconnectDelay(instanceID string, failures int) time.Duration { + steps := [...]time.Duration{2 * time.Second, 5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute} + if failures < 0 { + failures = 0 + } + if failures >= len(steps) { + failures = len(steps) - 1 + } + base := steps[failures] + h := fnv.New32a() + _, _ = h.Write([]byte(instanceID)) + jitterWindow := base / 5 + if jitterWindow <= 0 { + return base + } + return base + time.Duration(h.Sum32()%uint32(jitterWindow)) +} + +func canAutoReconnect(jid string) bool { + return strings.TrimSpace(jid) != "" +} + +func (w *whatsmeowService) reserveReconnect(instanceID string) (time.Duration, bool) { + w.reconnectMu.Lock() + defer w.reconnectMu.Unlock() + + state := w.reconnectState[instanceID] + if state == nil { + state = &instanceReconnectState{} + w.reconnectState[instanceID] = state + } + if state.scheduled { + return 0, false + } + state.scheduled = true + return reconnectDelay(instanceID, state.failures), true +} + +func (w *whatsmeowService) finishReconnect(instanceID string, connected bool) { + w.reconnectMu.Lock() + defer w.reconnectMu.Unlock() + + state := w.reconnectState[instanceID] + if state == nil { + return + } + state.scheduled = false + if connected { + state.failures = 0 + } else if state.failures < 1000 { + state.failures++ + } } type MyClient struct { @@ -171,74 +242,109 @@ type ProxyConfig struct { Username string `json:"username"` } -func (w whatsmeowService) ReconnectClient(instanceId string) error { - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Starting reconnection process - simulating restart", instanceId) +func (w *whatsmeowService) ReconnectClient(instanceId string) error { + w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Starting controlled reconnection", instanceId) - // Passo 1: Limpar conexão existente se houver - if client, exists := w.clientPointer[instanceId]; exists { - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Disconnecting existing client", instanceId) + w.runtimeMu.Lock() + client := w.clientPointer[instanceId] + mycli := w.myClientPointer[instanceId] + killChan := w.killChannel[instanceId] + done := w.doneChannel[instanceId] + w.runtimeMu.Unlock() - // Desconectar o cliente WebSocket + if client != nil { + if mycli != nil && mycli.eventHandlerID != 0 { + client.RemoveEventHandler(mycli.eventHandlerID) + } if client.IsConnected() { client.Disconnect() - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] WebSocket disconnected", instanceId) - } - - // Remover event handler se existir - if mycli, ok := w.myClientPointer[instanceId]; ok { - if mycli.eventHandlerID != 0 { - client.RemoveEventHandler(mycli.eventHandlerID) - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Event handler removed", instanceId) - } } } - - // Passo 2: Limpar todos os recursos da instância - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Cleaning up resources", instanceId) - - // Enviar sinal de kill se o canal existir - if killChan, exists := w.killChannel[instanceId]; exists { + if killChan != nil { select { case killChan <- true: - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Kill signal sent", instanceId) default: - // Canal pode estar bloqueado, continua } } - - // Remover das estruturas - delete(w.clientPointer, instanceId) - delete(w.myClientPointer, instanceId) - delete(w.killChannel, instanceId) - - // Limpar cache de userInfo para esta instância - if instance, err := w.instanceRepository.GetInstanceByID(instanceId); err == nil { - w.userInfoCache.Delete(instance.Token) - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] UserInfo cache cleared for token: %s", instanceId, instance.Token) + if done != nil { + select { + case <-done: + case <-time.After(5 * time.Second): + return fmt.Errorf("timed out waiting for previous client cleanup") + } } - // Passo 3: Atualizar status no banco instance, err := w.instanceRepository.GetInstanceByID(instanceId) if err != nil { return fmt.Errorf("failed to get instance: %v", err) } - + w.userInfoCache.Delete(instance.Token) instance.Connected = false instance.DisconnectReason = "Reconnecting" - err = w.instanceRepository.UpdateConnected(instanceId, false, "Reconnecting") - if err != nil { + if err = w.instanceRepository.UpdateConnected(instanceId, false, "Reconnecting"); err != nil { w.loggerWrapper.GetLogger(instanceId).LogWarn("[%s] Failed to update disconnect status: %v", instanceId, err) } - // Passo 4: Aguardar um pouco para garantir limpeza completa - time.Sleep(2 * time.Second) - - // Passo 5: Iniciar nova instância como se fosse a primeira vez w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Starting fresh instance", instanceId) return w.StartInstance(instanceId) } -func (w whatsmeowService) ForceUpdateJid(instanceId string, number string) error { +func (w *whatsmeowService) ScheduleReconnect(instanceId string) { + instance, err := w.instanceRepository.GetInstanceByID(instanceId) + if err != nil { + w.loggerWrapper.GetLogger(instanceId).LogWarn("[%s] Auto-reconnect skipped: instance lookup failed", instanceId) + return + } + if !canAutoReconnect(instance.Jid) { + w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Auto-reconnect suppressed for unpaired instance; explicit connect is required", instanceId) + return + } + + delay, reserved := w.reserveReconnect(instanceId) + if !reserved { + return + } + + go func() { + timer := time.NewTimer(delay) + defer timer.Stop() + <-timer.C + + w.runtimeMu.Lock() + client := w.clientPointer[instanceId] + connected := client != nil && client.IsConnected() && client.IsLoggedIn() + w.runtimeMu.Unlock() + if !connected { + if err := w.ReconnectClient(instanceId); err != nil { + w.loggerWrapper.GetLogger(instanceId).LogWarn("[%s] Controlled reconnect attempt failed: %v", instanceId, err) + } + } + + deadline := time.Now().Add(reconnectProbeTimeout) + for !connected && time.Now().Before(deadline) { + time.Sleep(time.Second) + w.runtimeMu.Lock() + client = w.clientPointer[instanceId] + connected = client != nil && client.IsConnected() && client.IsLoggedIn() + w.runtimeMu.Unlock() + } + + w.finishReconnect(instanceId, connected) + if !connected { + w.ScheduleReconnect(instanceId) + } + }() +} + +func (w *whatsmeowService) markReconnectSuccess(instanceId string) { + w.reconnectMu.Lock() + if state := w.reconnectState[instanceId]; state != nil { + state.failures = 0 + } + w.reconnectMu.Unlock() +} + +func (w *whatsmeowService) ForceUpdateJid(instanceId string, number string) error { instance, err := w.instanceRepository.GetInstanceByID(instanceId) if err != nil { w.loggerWrapper.GetLogger(instanceId).LogError("[%s] Error getting instance: %v", instanceId, err) @@ -301,7 +407,22 @@ func (w whatsmeowService) ForceUpdateJid(instanceId string, number string) error return nil } -func (w whatsmeowService) StartClient(cd *ClientData) { +func (w *whatsmeowService) deviceContainer(dbLog waLog.Logger) (*sqlstore.Container, bool, error) { + if w.config.PostgresAuthDB != "" { + w.authContainerOnce.Do(func() { + w.authContainer = sqlstore.NewWithDB(w.authDB, "postgres", dbLog) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + w.authContainerErr = w.authContainer.Upgrade(ctx) + }) + return w.authContainer, false, w.authContainerErr + } + dsn := fmt.Sprintf("file:%s/dbdata/main.db?_pragma=foreign_keys(1)&_busy_timeout=5000&cache=shared&mode=rwc&_journal_mode=WAL", w.exPath) + container, err := sqlstore.New(context.Background(), "sqlite", dsn, dbLog) + return container, true, err +} + +func (w *whatsmeowService) StartClient(cd *ClientData) { w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("Starting websocket connection to Whatsapp for user '%s'", cd.Instance.Id) @@ -314,29 +435,23 @@ func (w whatsmeowService) StartClient(cd *ClientData) { } } - var container *sqlstore.Container - + var dbLog waLog.Logger if w.config.WaDebug != "" { - dbLog := waLog.Stdout("Database", w.config.WaDebug, true) - if w.config.PostgresAuthDB != "" { - container, err = sqlstore.New(context.Background(), "postgres", w.config.PostgresAuthDB, dbLog) - } else { - dsn := fmt.Sprintf("file:%s/dbdata/main.db?_pragma=foreign_keys(1)&_busy_timeout=5000&cache=shared&mode=rwc&_journal_mode=WAL", w.exPath) - container, err = sqlstore.New(context.Background(), "sqlite", dsn, dbLog) - } - } else { - if w.config.PostgresAuthDB != "" { - container, err = sqlstore.New(context.Background(), "postgres", w.config.PostgresAuthDB, nil) - } else { - dsn := fmt.Sprintf("file:%s/dbdata/main.db?_pragma=foreign_keys(1)&_busy_timeout=5000&cache=shared&mode=rwc&_journal_mode=WAL", w.exPath) - container, err = sqlstore.New(context.Background(), "sqlite", dsn, nil) - } + dbLog = waLog.Stdout("Database", w.config.WaDebug, true) } + container, ownedContainer, err := w.deviceContainer(dbLog) if err != nil { w.loggerWrapper.GetLogger(cd.Instance.Id).LogError("[%s] Failed to create container: %v", cd.Instance.Id, err) return } + if ownedContainer { + defer func() { + if err := container.Close(); err != nil { + w.loggerWrapper.GetLogger(cd.Instance.Id).LogWarn("[%s] Failed to close device store: %v", cd.Instance.Id, err) + } + }() + } if cd.Instance.Jid != "" { jid, _ := utils.ParseJID(cd.Instance.Jid) @@ -412,7 +527,29 @@ func (w whatsmeowService) StartClient(cd *ClientData) { clientLog := waLog.Stdout("Client", minLevel, true) client := whatsmeow.NewClient(deviceStore, clientLog) + w.runtimeMu.Lock() + killChan := w.killChannel[cd.Instance.Id] + if killChan == nil { + killChan = make(chan bool, 1) + w.killChannel[cd.Instance.Id] = killChan + } + done := make(chan struct{}) + w.doneChannel[cd.Instance.Id] = done w.clientPointer[cd.Instance.Id] = client + w.runtimeMu.Unlock() + defer func() { + w.runtimeMu.Lock() + if w.clientPointer[cd.Instance.Id] == client { + delete(w.clientPointer, cd.Instance.Id) + delete(w.myClientPointer, cd.Instance.Id) + delete(w.killChannel, cd.Instance.Id) + } + if w.doneChannel[cd.Instance.Id] == done { + delete(w.doneChannel, cd.Instance.Id) + } + close(done) + w.runtimeMu.Unlock() + }() if cd.IsProxy { var proxyConfig ProxyConfig @@ -465,7 +602,7 @@ func (w whatsmeowService) StartClient(cd *ClientData) { client.AutoTrustIdentity = true mycli := &MyClient{ - service: &w, + service: w, Instance: cd.Instance, WAClient: client, eventHandlerID: 1, @@ -500,7 +637,9 @@ func (w whatsmeowService) StartClient(cd *ClientData) { mycli.eventHandlerID = mycli.WAClient.AddEventHandler(mycli.myEventHandler) // Armazena o MyClient no map para permitir atualizações posteriores + w.runtimeMu.Lock() w.myClientPointer[cd.Instance.Id] = mycli + w.runtimeMu.Unlock() if client.Store.ID != nil { w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("[%s] Already logged in with JID: %s", cd.Instance.Id, client.Store.ID.String()) @@ -570,66 +709,31 @@ func (w whatsmeowService) StartClient(cd *ClientData) { } - // Removed auto-reconnect logic to prevent infinite loops - - for { - select { - case <-w.killChannel[cd.Instance.Id]: - w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("Received kill signal for user '%s'", cd.Instance.Id) - client.Disconnect() - - delete(w.clientPointer, cd.Instance.Id) - delete(w.myClientPointer, cd.Instance.Id) - - // Limpar cache de userInfo para esta instância - w.userInfoCache.Delete(cd.Instance.Token) - w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("[%s] UserInfo cache cleared for token: %s", cd.Instance.Id, cd.Instance.Token) - - cd.Instance.Connected = false - - err := w.instanceRepository.UpdateConnected(cd.Instance.Id, cd.Instance.Connected, cd.Instance.DisconnectReason) - if err != nil { - w.loggerWrapper.GetLogger(cd.Instance.Id).LogError("[%s] Error updating instance: %s", cd.Instance.Id, err) - } - - postMap := make(map[string]interface{}) - - postMap["event"] = "LoggedOut" - - dataMap := make(map[string]interface{}) - - dataMap["reason"] = "Logged out" - - postMap["data"] = dataMap - - postMap["instanceToken"] = mycli.token - postMap["instanceId"] = mycli.userID - postMap["instanceName"] = cd.Instance.Name - - var queueName string - - if _, ok := postMap["event"]; ok { - queueName = strings.ToLower(fmt.Sprintf("%s.%s", cd.Instance.Id, postMap["event"])) - } - - values, err := json.Marshal(postMap) - if err != nil { - w.loggerWrapper.GetLogger(cd.Instance.Id).LogError("[%s] Failed to marshal JSON for queue", cd.Instance.Id) - return - } - - go w.CallWebhook(cd.Instance, queueName, values) - - if mycli.config.AmqpGlobalEnabled || mycli.config.NatsGlobalEnabled { - go mycli.service.SendToGlobalQueues(postMap["event"].(string), values, mycli.userID) - } + // A kill signal always means stop this runtime. Reconnection is scheduled + // separately and serialized per instance; never recurse from cleanup. + <-killChan + w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("Received kill signal for user '%s'", cd.Instance.Id) + if client.IsConnected() { + client.Disconnect() + } + w.userInfoCache.Delete(cd.Instance.Token) + cd.Instance.Connected = false + if err := w.instanceRepository.UpdateConnected(cd.Instance.Id, false, cd.Instance.DisconnectReason); err != nil { + w.loggerWrapper.GetLogger(cd.Instance.Id).LogError("[%s] Error updating instance: %s", cd.Instance.Id, err) + } - // restart client - w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("[%s] Restarting client", cd.Instance.Id) - w.StartClient(cd) - return - default: - time.Sleep(1000 * time.Millisecond) + postMap := map[string]interface{}{ + "event": "LoggedOut", + "data": map[string]interface{}{"reason": "Runtime stopped"}, + "instanceToken": mycli.token, + "instanceId": mycli.userID, + "instanceName": cd.Instance.Name, + } + if values, err := json.Marshal(postMap); err == nil { + queueName := strings.ToLower(fmt.Sprintf("%s.%s", cd.Instance.Id, postMap["event"])) + go w.CallWebhook(cd.Instance, queueName, values) + if mycli.config.AmqpGlobalEnabled || mycli.config.NatsGlobalEnabled { + go mycli.service.SendToGlobalQueues(postMap["event"].(string), values, mycli.userID) } } } @@ -732,14 +836,8 @@ func (mycli *MyClient) handleQRCodes(codes []string) { mycli.loggerWrapper.GetLogger(instanceID).LogInfo("[%s] QR max-count reached but passkey ceremony active — not tearing down", instanceID) return } - mycli.loggerWrapper.GetLogger(instanceID).LogWarn("[%s] Maximum QR code count reached (%d), forcing logout and QRTimeout", instanceID, mycli.config.QrcodeMaxCount) - - if mycli.WAClient.IsConnected() { - if err := mycli.WAClient.Logout(context.Background()); err != nil { - mycli.loggerWrapper.GetLogger(instanceID).LogWarn("[%s] Error during forced logout: %v", instanceID, err) - } - } - mycli.teardownQR(fmt.Sprintf("Maximum QR code count (%d) reached", mycli.config.QrcodeMaxCount), true) + mycli.loggerWrapper.GetLogger(instanceID).LogWarn("[%s] Maximum QR code count reached (%d); stopping unpaired runtime until an explicit connect request", instanceID, mycli.config.QrcodeMaxCount) + mycli.teardownQR(fmt.Sprintf("Maximum QR code count (%d) reached", mycli.config.QrcodeMaxCount), false) return } @@ -844,12 +942,14 @@ func (mycli *MyClient) teardownQR(reason string, forceLogout bool) { } } - // Signal StartClient's select loop to disconnect and clean up the shared - // maps (it is the single writer for this instance). Blocking send mirrors - // the original timeout branch so the signal is never dropped. + // Signal StartClient to stop. The buffered channel makes this idempotent and + // avoids a QR goroutine blocking forever when cleanup is already pending. mycli.loggerWrapper.GetLogger(instanceID).LogWarn("[%s] QR timeout — signaling kill channel", instanceID) if killChan, exists := mycli.killChannel[instanceID]; exists { - killChan <- true + select { + case killChan <- true: + default: + } } } @@ -875,6 +975,7 @@ func (mycli *MyClient) myEventHandler(rawEvt interface{}) { } } case *events.Connected, *events.PushNameSetting: + mycli.service.markReconnectSuccess(mycli.userID) mycli.loggerWrapper.GetLogger(mycli.userID).LogInfo("[%s] events.Connected to Whatsapp for user '%s'", mycli.userID, mycli.WAClient.Store.PushName) if len(mycli.WAClient.Store.PushName) > 0 { doWebhook = true @@ -1981,13 +2082,10 @@ func (mycli *MyClient) myEventHandler(rawEvt interface{}) { mycli.loggerWrapper.GetLogger(mycli.userID).LogError("[%s] Error updating instance: %s", mycli.Instance.Id, err) } - // Trigger instance restart via websocket-capable service (non-blocking) - go func(instanceID string) { - mycli.loggerWrapper.GetLogger(instanceID).LogInfo("[%s] Disconnected detected, restarting instance", instanceID) - if err := mycli.service.ReconnectClient(instanceID); err != nil { - mycli.loggerWrapper.GetLogger(instanceID).LogError("[%s] Failed to restart instance: %v", instanceID, err) - } - }(mycli.userID) + // Paired instances reconnect through a per-instance scheduler with + // deduplication, exponential backoff and jitter. Unpaired instances are + // deliberately left stopped so QR generation cannot loop forever. + mycli.service.ScheduleReconnect(mycli.userID) case *events.LabelEdit: doWebhook = true postMap["event"] = "LabelEdit" @@ -2320,7 +2418,7 @@ func (w *whatsmeowService) sendToQueueOrWebhook(instance *instance_model.Instanc } } -func (w whatsmeowService) StartInstance(instanceId string) error { +func (w *whatsmeowService) StartInstance(instanceId string) error { instance, err := w.instanceRepository.GetInstanceByID(instanceId) if err != nil { return err @@ -2382,7 +2480,9 @@ func (w whatsmeowService) StartInstance(instanceId string) error { } } - w.killChannel[instance.Id] = make(chan bool) + w.runtimeMu.Lock() + w.killChannel[instance.Id] = make(chan bool, 1) + w.runtimeMu.Unlock() clientData := &ClientData{ Instance: instance, @@ -2409,7 +2509,7 @@ func (w whatsmeowService) StartInstance(instanceId string) error { return nil } -func (w whatsmeowService) ConnectOnStartup(clientName string) { +func (w *whatsmeowService) ConnectOnStartup(clientName string) { w.loggerWrapper.GetLogger(clientName).LogInfo("Connecting all instances on startup") var instances []*instance_model.Instance var err error @@ -2672,7 +2772,7 @@ func fetchWhatsAppWebVersion() (*clientVersion, error) { return nil, fmt.Errorf("could not find client revision in the fetched content. Content preview: %s", content) } -func (w whatsmeowService) UpdateInstanceSettings(instanceId string) error { +func (w *whatsmeowService) UpdateInstanceSettings(instanceId string) error { w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Updating instance settings in runtime", instanceId) // Busca a instância atualizada do banco @@ -2731,7 +2831,7 @@ func (w whatsmeowService) UpdateInstanceSettings(instanceId string) error { return nil } -func (w whatsmeowService) UpdateInstanceAdvancedSettings(instanceId string) error { +func (w *whatsmeowService) UpdateInstanceAdvancedSettings(instanceId string) error { w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Updating advanced settings in runtime", instanceId) // Busca a instância atualizada do banco @@ -2755,38 +2855,28 @@ func (w whatsmeowService) UpdateInstanceAdvancedSettings(instanceId string) erro return nil } -func (w whatsmeowService) ClearInstanceCache(instanceId string, token string) error { - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Clearing instance cache - Token: %s", instanceId, token) - - // Limpar userInfoCache +func (w *whatsmeowService) ClearInstanceCache(instanceId string, token string) error { + w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Clearing instance runtime", instanceId) w.userInfoCache.Delete(token) - // Limpar myClientPointer se existir - if _, exists := w.myClientPointer[instanceId]; exists { - delete(w.myClientPointer, instanceId) - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] MyClient pointer cleared", instanceId) - } - - // Limpar clientPointer se existir - if _, exists := w.clientPointer[instanceId]; exists { - delete(w.clientPointer, instanceId) - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Client pointer cleared", instanceId) - } - - // Limpar killChannel se existir - if killChan, exists := w.killChannel[instanceId]; exists { + w.runtimeMu.Lock() + killChan := w.killChannel[instanceId] + done := w.doneChannel[instanceId] + w.runtimeMu.Unlock() + if killChan != nil { select { case killChan <- true: - // Canal recebeu o sinal default: - // Canal pode estar bloqueado, apenas fecha } - close(killChan) - delete(w.killChannel, instanceId) - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Kill channel cleared", instanceId) } - - w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Instance cache completely cleared", instanceId) + if done != nil { + select { + case <-done: + case <-time.After(5 * time.Second): + return fmt.Errorf("timed out waiting for instance runtime cleanup") + } + } + w.loggerWrapper.GetLogger(instanceId).LogInfo("[%s] Instance runtime cleared", instanceId) return nil } @@ -2831,6 +2921,8 @@ func NewWhatsmeowService( natsProducer: natsProducer, loggerWrapper: loggerWrapper, passkeyCeremony: ceremony.NewStore(), + doneChannel: make(map[string]chan struct{}), + reconnectState: make(map[string]*instanceReconnectState), } } From 3a2dcc79d5a57c4f4c759420a7dfc85dec818e15 Mon Sep 17 00:00:00 2001 From: LembreteAI Build Date: Wed, 29 Jul 2026 13:44:38 -0300 Subject: [PATCH 2/2] fix: never start a second client for the same instance A QR request could spawn a runtime for an instance that already had one: GetQr started a client whenever clientPointer was nil or the client was logged in. Two clients on the same instance share one killChannel and one database row, so when the spare one exhausted its QR cycle it tore the runtime down and flagged the instance as disconnected while the paired session stayed online. Callers polling for a QR then kept feeding the loop. StartClient now admits a single runtime per instance, guarded by a token so a late cleanup cannot free the reservation of its replacement, and GetQr refuses to restart a logged-in session instead of racing it. --- pkg/instance/service/instance_service.go | 41 +++++++++---- pkg/instance/service/qr_runtime_test.go | 32 ++++++++++ pkg/whatsmeow/service/runtime_test.go | 78 ++++++++++++++++++++++++ pkg/whatsmeow/service/whatsmeow.go | 54 ++++++++++++++-- 4 files changed, 188 insertions(+), 17 deletions(-) create mode 100644 pkg/instance/service/qr_runtime_test.go create mode 100644 pkg/whatsmeow/service/runtime_test.go diff --git a/pkg/instance/service/instance_service.go b/pkg/instance/service/instance_service.go index 6d40ce70..8a8dc588 100644 --- a/pkg/instance/service/instance_service.go +++ b/pkg/instance/service/instance_service.go @@ -411,21 +411,40 @@ func (i instances) Status(instance *instance_model.Instance) (*StatusStruct, err }, nil } +// ErrSessionAlreadyLoggedIn is returned when a QR code is requested for an +// instance that is already paired and online. +var ErrSessionAlreadyLoggedIn = errors.New("session already logged in") + +// qrNeedsNewRuntime decides whether a QR request may start a client. +// +// A paired, logged-in instance must never get a second runtime: the two clients +// share a killChannel, so the spare one exhausts its QR cycle and tears down the +// live session, which then shows up as a disconnected instance holding an open +// WhatsApp socket. A client that exists but is not logged in is already in its +// QR cycle and is reused. +func qrNeedsNewRuntime(clientExists, loggedIn bool) (bool, error) { + if loggedIn { + return false, ErrSessionAlreadyLoggedIn + } + + return !clientExists, nil +} + func (i instances) GetQr(instance *instance_model.Instance) (*QrcodeStruct, error) { logger := i.loggerWrapper.GetLogger(instance.Id) client := i.clientPointer[instance.Id] - // Se não há cliente ou o cliente está logado, precisamos iniciar um novo cliente - if client == nil || client.IsLoggedIn() { - if client != nil && client.IsLoggedIn() { - logger.LogInfo("[%s] Client is logged in, starting new instance for QR code", instance.Id) - } else { - logger.LogInfo("[%s] No client found, starting new instance for QR code", instance.Id) - } + startRuntime, err := qrNeedsNewRuntime(client != nil, client != nil && client.IsLoggedIn()) + if err != nil { + logger.LogInfo("[%s] QR requested for a logged-in session; keeping the current runtime", instance.Id) + return nil, err + } + + if startRuntime { + logger.LogInfo("[%s] No client found, starting new instance for QR code", instance.Id) // Iniciar nova instância para gerar QR code - err := i.whatsmeowService.StartInstance(instance.Id) - if err != nil { + if err = i.whatsmeowService.StartInstance(instance.Id); err != nil { logger.LogError("[%s] Failed to start instance: %v", instance.Id, err) return nil, fmt.Errorf("failed to start instance: %w", err) } @@ -437,7 +456,7 @@ func (i instances) GetQr(instance *instance_model.Instance) (*QrcodeStruct, erro // Verificar novamente se há cliente client = i.clientPointer[instance.Id] if client != nil && client.IsLoggedIn() { - return nil, fmt.Errorf("session already logged in") + return nil, ErrSessionAlreadyLoggedIn } } else if !client.IsConnected() { // Se o cliente existe mas não está conectado, pode estar aguardando QR code @@ -445,7 +464,7 @@ func (i instances) GetQr(instance *instance_model.Instance) (*QrcodeStruct, erro } // Buscar instância atualizada do banco para pegar o QR code mais recente - instance, err := i.instanceRepository.GetInstanceByID(instance.Id) + instance, err = i.instanceRepository.GetInstanceByID(instance.Id) if err != nil { return nil, err } diff --git a/pkg/instance/service/qr_runtime_test.go b/pkg/instance/service/qr_runtime_test.go new file mode 100644 index 00000000..8e9a6501 --- /dev/null +++ b/pkg/instance/service/qr_runtime_test.go @@ -0,0 +1,32 @@ +package instance_service + +import ( + "errors" + "testing" +) + +func TestQrNeedsNewRuntime(t *testing.T) { + cases := []struct { + name string + exists bool + loggedIn bool + wantStart bool + wantErrorIs error + }{ + {name: "no client at all starts one", exists: false, wantStart: true}, + {name: "client waiting for QR reuses it", exists: true, wantStart: false}, + {name: "logged in client is never restarted", exists: true, loggedIn: true, wantErrorIs: ErrSessionAlreadyLoggedIn}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + start, err := qrNeedsNewRuntime(tc.exists, tc.loggedIn) + if !errors.Is(err, tc.wantErrorIs) { + t.Fatalf("error = %v, want %v", err, tc.wantErrorIs) + } + if start != tc.wantStart { + t.Fatalf("start = %v, want %v", start, tc.wantStart) + } + }) + } +} diff --git a/pkg/whatsmeow/service/runtime_test.go b/pkg/whatsmeow/service/runtime_test.go new file mode 100644 index 00000000..af31ee00 --- /dev/null +++ b/pkg/whatsmeow/service/runtime_test.go @@ -0,0 +1,78 @@ +package whatsmeow_service + +import ( + "sync" + "sync/atomic" + "testing" +) + +func TestReserveRuntimeAllowsOneRuntimePerInstance(t *testing.T) { + service := &whatsmeowService{runtimeToken: make(map[string]uint64)} + + var accepted atomic.Int32 + var wg sync.WaitGroup + for attempt := 0; attempt < 100; attempt++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, ok := service.reserveRuntime("tenant-7"); ok { + accepted.Add(1) + } + }() + } + wg.Wait() + + if got := accepted.Load(); got != 1 { + t.Fatalf("expected exactly one runtime per instance, got %d", got) + } +} + +func TestReserveRuntimeIsPerInstance(t *testing.T) { + service := &whatsmeowService{runtimeToken: make(map[string]uint64)} + + if _, ok := service.reserveRuntime("tenant-a"); !ok { + t.Fatal("first instance must be able to start") + } + if _, ok := service.reserveRuntime("tenant-b"); !ok { + t.Fatal("a different instance must not be blocked by another one's runtime") + } +} + +func TestReleaseRuntimeAllowsRestart(t *testing.T) { + service := &whatsmeowService{runtimeToken: make(map[string]uint64)} + + token, ok := service.reserveRuntime("tenant-7") + if !ok { + t.Fatal("first reservation must succeed") + } + service.releaseRuntime("tenant-7", token) + + if _, ok := service.reserveRuntime("tenant-7"); !ok { + t.Fatal("a new runtime must start once the previous one is released") + } +} + +// A late cleanup from a dead runtime must not free the reservation of the one +// that took its place — that is how a second concurrent client gets in. +func TestReleaseRuntimeIgnoresStaleToken(t *testing.T) { + service := &whatsmeowService{runtimeToken: make(map[string]uint64)} + + stale, _ := service.reserveRuntime("tenant-7") + service.releaseRuntime("tenant-7", stale) + + current, ok := service.reserveRuntime("tenant-7") + if !ok { + t.Fatal("replacement runtime must be able to reserve") + } + + service.releaseRuntime("tenant-7", stale) + + if _, ok := service.reserveRuntime("tenant-7"); ok { + t.Fatal("stale release freed the live runtime's reservation") + } + + service.releaseRuntime("tenant-7", current) + if _, ok := service.reserveRuntime("tenant-7"); !ok { + t.Fatal("current token must still release the reservation") + } +} diff --git a/pkg/whatsmeow/service/whatsmeow.go b/pkg/whatsmeow/service/whatsmeow.go index 22b68b24..54dbe45a 100644 --- a/pkg/whatsmeow/service/whatsmeow.go +++ b/pkg/whatsmeow/service/whatsmeow.go @@ -102,6 +102,9 @@ type whatsmeowService struct { passkeyCeremony *ceremony.Store runtimeMu sync.Mutex doneChannel map[string]chan struct{} + reserveMu sync.Mutex + runtimeToken map[string]uint64 + runtimeCounter uint64 reconnectMu sync.Mutex reconnectState map[string]*instanceReconnectState authContainerOnce sync.Once @@ -138,6 +141,38 @@ func canAutoReconnect(jid string) bool { return strings.TrimSpace(jid) != "" } +// reserveRuntime admits a single live runtime per instance. +// +// Two clients on the same instance share one killChannel and one row in the +// database: whichever exhausts its QR cycle first tears the runtime down and +// flags the instance as disconnected, even when the other one is paired and +// online. Connect, ForceReconnect, StartInstance and the QR endpoint all reach +// StartClient, so the guard belongs here rather than at each call site. +// +// The token makes release safe: a cleanup arriving late cannot free the +// reservation of the runtime that already replaced it. +func (w *whatsmeowService) reserveRuntime(instanceID string) (uint64, bool) { + w.reserveMu.Lock() + defer w.reserveMu.Unlock() + + if _, active := w.runtimeToken[instanceID]; active { + return 0, false + } + w.runtimeCounter++ + w.runtimeToken[instanceID] = w.runtimeCounter + + return w.runtimeCounter, true +} + +func (w *whatsmeowService) releaseRuntime(instanceID string, token uint64) { + w.reserveMu.Lock() + defer w.reserveMu.Unlock() + + if w.runtimeToken[instanceID] == token { + delete(w.runtimeToken, instanceID) + } +} + func (w *whatsmeowService) reserveReconnect(instanceID string) (time.Duration, bool) { w.reconnectMu.Lock() defer w.reconnectMu.Unlock() @@ -424,17 +459,20 @@ func (w *whatsmeowService) deviceContainer(dbLog waLog.Logger) (*sqlstore.Contai func (w *whatsmeowService) StartClient(cd *ClientData) { + runtimeToken, reserved := w.reserveRuntime(cd.Instance.Id) + if !reserved { + w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("[%s] Runtime already active; not starting a second client", cd.Instance.Id) + return + } + // Covers the paths that return before the cleanup defer below is registered. + // Releasing twice is harmless: the token no longer matches. + defer w.releaseRuntime(cd.Instance.Id, runtimeToken) + w.loggerWrapper.GetLogger(cd.Instance.Id).LogInfo("Starting websocket connection to Whatsapp for user '%s'", cd.Instance.Id) var deviceStore *store.Device var err error - if w.clientPointer[cd.Instance.Id] != nil { - if w.clientPointer[cd.Instance.Id].IsConnected() { - return - } - } - var dbLog waLog.Logger if w.config.WaDebug != "" { dbLog = waLog.Stdout("Database", w.config.WaDebug, true) @@ -547,6 +585,9 @@ func (w *whatsmeowService) StartClient(cd *ClientData) { if w.doneChannel[cd.Instance.Id] == done { delete(w.doneChannel, cd.Instance.Id) } + // Before closing `done`: ReconnectClient waits on it and then starts the + // replacement runtime, which needs the reservation to be free already. + w.releaseRuntime(cd.Instance.Id, runtimeToken) close(done) w.runtimeMu.Unlock() }() @@ -2922,6 +2963,7 @@ func NewWhatsmeowService( loggerWrapper: loggerWrapper, passkeyCeremony: ceremony.NewStore(), doneChannel: make(map[string]chan struct{}), + runtimeToken: make(map[string]uint64), reconnectState: make(map[string]*instanceReconnectState), } }