Defer server AB props acknowledgements until sync completion

client.go

@@ -157,7 +157,13 @@ type Client struct {
 	responseWaitersLock sync.Mutex
 	businessCatalogAuth atomic.Pointer[businessCatalogAuthState]
 
-	handlerQueue      chan *waBinary.Node
+	handlerQueue chan *waBinary.Node
+	// nodeHandlers maps a top-level stanza tag to its handler. Upstream
+	// whatsmeow exposes the same unexported map; keeping it (instead of a
+	// closed switch) lets a headless shadow Client dispatch injected nodes
+	// synchronously and lets embedders that reach for it via reflection keep
+	// working.
+	nodeHandlers      map[string]nodeHandler
 	eventHandlers     []wrappedEventHandler
 	eventHandlersLock sync.RWMutex
 
@@ -226,6 +232,19 @@ type Client struct {
 	// after decoding but before standard dispatch. See [RawNodeHandler].
 	RawNodeHandler RawNodeHandler
 
+	// ServerABPropsSync optionally completes a server AB-props task before its
+	// acknowledgement. Install before Connect. A nil handler preserves ordinary
+	// dispatch. A returned error withholds the ack; an ordinary unsuccessful sync
+	// may return nil after its task has completed. The receive loop remains free
+	// to process the IQ response. The handler must honor context cancellation.
+	ServerABPropsSync    func(context.Context) error
+	serverABPropsMu      sync.Mutex
+	serverABPropsPending map[string]struct{}
+	// shadowRelay, if non-nil, marks this client as a headless "shadow"
+	// client (see [NewShadowClient]): it has no socket, Connect is guarded,
+	// and outbound nodes plus Signal/keying oracle ops are delegated here.
+	shadowRelay ShadowRelay
+
 	// DisabledFeatures controls which built-in processing paths are
 	// skipped. See [DisabledFeatures].
 	DisabledFeatures DisabledFeatures
@@ -345,6 +364,7 @@ func NewClient(deviceStore *store.Device, log waLog.Logger) *Client {
 		WebSocketHeaders: http.Header{},
 	}
 	cli.paired.Store(deviceStore.ID != nil)
+	cli.nodeHandlers = cli.defaultNodeHandlers()
 	return cli
 }
 
@@ -585,6 +605,12 @@ func (cli *Client) connect(ctx context.Context) error {
 }
 
 func (cli *Client) unlockedConnect(ctx context.Context) error {
+	if cli.shadowRelay != nil {
+		// Guard: a shadow client must never open a socket. All Connect
+		// paths (Connect, ConnectContext, autoReconnect) funnel through
+		// here, so failing closed here blocks every one of them.
+		return ErrShadowClientNoConnect
+	}
 	if cli.Store.Deleted {
 		return store.ErrDeviceDeleted
 	}
@@ -995,40 +1021,34 @@ Loop:
 }
 
 func (cli *Client) hasNodeHandler(tag string) bool {
-	switch tag {
-	case "message", "status", "appdata", "receipt", "call", "chatstate", "presence", "notification", "success", "failure", "stream:error", "iq", "ib":
-		return true
-	default:
-		return false
+	_, ok := cli.nodeHandlers[tag]
+	return ok
+}
+
+type nodeHandler func(ctx context.Context, node *waBinary.Node)
+
+// defaultNodeHandlers returns the built-in top-level stanza dispatch table.
+func (cli *Client) defaultNodeHandlers() map[string]nodeHandler {
+	return map[string]nodeHandler{
+		"message":      cli.handleEncryptedMessage,
+		"appdata":      cli.handleEncryptedMessage,
+		"status":       cli.handleUnencryptedMessage,
+		"receipt":      cli.handleReceipt,
+		"call":         cli.handleCallEvent,
+		"chatstate":    cli.handleChatState,
+		"presence":     cli.handlePresence,
+		"notification": cli.handleNotification,
+		"success":      cli.handleConnectSuccess,
+		"failure":      cli.handleConnectFailure,
+		"stream:error": cli.handleStreamError,
+		"iq":           cli.handleIQ,
+		"ib":           cli.handleIB,
 	}
 }
 
 func (cli *Client) handleNode(ctx context.Context, node *waBinary.Node) {
-	switch node.Tag {
-	case "message", "appdata":
-		cli.handleEncryptedMessage(ctx, node)
-	case "status":
-		cli.handleUnencryptedMessage(ctx, node)
-	case "receipt":
-		cli.handleReceipt(ctx, node)
-	case "call":
-		cli.handleCallEvent(ctx, node)
-	case "chatstate":
-		cli.handleChatState(ctx, node)
-	case "presence":
-		cli.handlePresence(ctx, node)
-	case "notification":
-		cli.handleNotification(ctx, node)
-	case "success":
-		cli.handleConnectSuccess(ctx, node)
-	case "failure":
-		cli.handleConnectFailure(ctx, node)
-	case "stream:error":
-		cli.handleStreamError(ctx, node)
-	case "iq":
-		cli.handleIQ(ctx, node)
-	case "ib":
-		cli.handleIB(ctx, node)
+	if handler, ok := cli.nodeHandlers[node.Tag]; ok {
+		handler(ctx, node)
 	}
 }
 
@@ -1040,6 +1060,18 @@ func (cli *Client) sendNodeAndGetData(ctx context.Context, node waBinary.Node) (
 	sock := cli.socket
 	cli.socketLock.RUnlock()
 	if sock == nil {
+		// A headless (shadow) client has no socket by design; route the
+		// outbound node to its relay instead. Fail closed if there is
+		// neither a socket nor a relay so a write can never silently
+		// escape or nil-panic on the absent socket.
+		if cli.shadowRelay != nil {
+			payload, err := waBinary.Marshal(node)
+			if err != nil {
+				return nil, fmt.Errorf("failed to marshal node: %w", err)
+			}
+			cli.sendLog.Debugf("%s", &node)
+			return payload, cli.shadowRelay.SendNode(ctx, payload)
+		}
 		return nil, ErrNotConnected
 	}
 

message.go

@@ -437,6 +437,20 @@ func (cli *Client) decryptMessages(ctx context.Context, info *types.MessageInfo,
 		} else if errors.Is(err, signalerror.ErrOldCounter) {
 			cli.Log.Warnf("Ignoring message %s from %s: %v", info.ID, info.SourceString(), err)
 			continue
+		} else if errors.Is(err, ErrShadowGroupUnsupported) {
+			// A shadow can never make group ciphertext decryptable, so a retry
+			// receipt would only provoke redeliveries. Acknowledge, surface the
+			// message as undecryptable, and move on.
+			cli.Log.Warnf("Ignoring group message %s from %s: %v", info.ID, info.SourceString(), err)
+			cli.backgroundIfAsyncAck(func() {
+				cli.sendAck(ctx, node, 0)
+			})
+			cli.dispatchEvent(&events.UndecryptableMessage{
+				Info:            *info,
+				IsUnavailable:   true,
+				DecryptFailMode: events.DecryptFailMode(ag.OptionalString("decrypt-fail")),
+			})
+			continue
 		} else if err != nil {
 			cli.Log.Warnf("Error decrypting message %s from %s: %v", info.ID, info.SourceString(), err)
 			if ctx.Err() != nil || errors.Is(err, context.Canceled) {
@@ -602,6 +616,11 @@ func (cli *Client) bufferedDecrypt(
 }
 
 func (cli *Client) decryptDM(ctx context.Context, child *waBinary.Node, from types.JID, isPreKey bool, serverTS time.Time) ([]byte, *[32]byte, error) {
+	if cli.isShadow() {
+		// A headless client holds no live Signal session; delegate
+		// decryption to the relay oracle.
+		return cli.shadowRelay.DecryptDM(ctx, child, from, isPreKey)
+	}
 	content, ok := child.Content.([]byte)
 	if !ok {
 		return nil, nil, fmt.Errorf("message content is not a byte slice")
@@ -652,6 +671,11 @@ func (cli *Client) decryptDM(ctx context.Context, child *waBinary.Node, from typ
 }
 
 func (cli *Client) decryptGroupMsg(ctx context.Context, child *waBinary.Node, from types.JID, chat types.JID, serverTS time.Time) ([]byte, *[32]byte, error) {
+	if cli.isShadow() {
+		// Sender-key state must never be created or read in a headless shadow;
+		// the relay oracle covers direct messages only. Fail closed.
+		return nil, nil, ErrShadowGroupUnsupported
+	}
 	content, ok := child.Content.([]byte)
 	if !ok {
 		return nil, nil, fmt.Errorf("message content is not a byte slice")
@@ -713,6 +737,9 @@ func padMessage(plaintext []byte) []byte {
 }
 
 func (cli *Client) handleSenderKeyDistributionMessage(ctx context.Context, chat, from types.JID, axolotlSKDM []byte) {
+	if cli.isShadow() {
+		return
+	}
 	builder := groups.NewGroupSessionBuilder(cli.Store, pbSerializer)
 	senderKeyName := protocol.NewSenderKeyName(chat.String(), from.SignalAddress())
 	sdkMsg, err := protocol.NewSenderKeyDistributionMessageFromBytes(axolotlSKDM, pbSerializer.SenderKeyDistributionMessage)

notification.go

@@ -507,6 +507,9 @@ func (cli *Client) handleStatusNotification(ctx context.Context, node *waBinary.
 }
 
 func (cli *Client) handleNotification(ctx context.Context, node *waBinary.Node) {
+	if cli.deferServerABProps(ctx, node, func() { cli.sendAck(ctx, node, 0) }) {
+		return
+	}
 	var cancelled bool
 	defer cli.maybeDeferredAck(ctx, node)(&cancelled)
 	ag := node.AttrGetter()
@@ -566,3 +569,52 @@ func (cli *Client) handleNotification(ctx context.Context, node *waBinary.Node)
 		cli.Log.Debugf("Unhandled notification with type %s", notifType)
 	}
 }
+
+// Only the server/abprops notification takes this path. In-flight duplicate
+// IDs share the task and ack. A later redelivery may start a fresh task.
+func (cli *Client) deferServerABProps(ctx context.Context, node *waBinary.Node, ack func()) bool {
+	if cli.ServerABPropsSync == nil || node == nil || node.Tag != "notification" || node.Attrs["type"] != "server" {
+		return false
+	}
+	if _, ok := node.GetOptionalChildByTag("abprops"); !ok {
+		return false
+	}
+	ag := node.AttrGetter()
+	id := ag.String("id")
+	from := ag.JID("from")
+	if !ag.OK() || id == "" || from.IsEmpty() {
+		return false
+	}
+	key := from.String() + "/" + id
+	cli.serverABPropsMu.Lock()
+	if cli.serverABPropsPending == nil {
+		cli.serverABPropsPending = make(map[string]struct{})
+	}
+	if _, exists := cli.serverABPropsPending[key]; exists {
+		cli.serverABPropsMu.Unlock()
+		return true
+	}
+	if len(cli.serverABPropsPending) >= 256 {
+		cli.serverABPropsMu.Unlock()
+		return true
+	}
+	cli.serverABPropsPending[key] = struct{}{}
+	cli.serverABPropsMu.Unlock()
+	go func() {
+		// Allow three 30-second attempts, two retry waits and durable completion.
+		taskCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+		defer cancel()
+		defer func() {
+			cli.serverABPropsMu.Lock()
+			delete(cli.serverABPropsPending, key)
+			cli.serverABPropsMu.Unlock()
+		}()
+		if ctx.Err() != nil {
+			return
+		}
+		if err := cli.ServerABPropsSync(taskCtx); err == nil && taskCtx.Err() == nil {
+			ack()
+		}
+	}()
+	return true
+}