[cli][coap] fix CoAP-observe messaging and add test (#12103)

This commit contains some improvements and fixes to the CoAP-observe (RFC 7641) messaging model implementation. It also adds an 'expect' test for using the CoAP-observe related CLI commands. Specific items: - ensure that a NON observe request is not acknowledged with an Ack. - enable a NON observe request to never time out, unless cancelled explicitly, or unless 0 observe responses are received within the NON request's timeout period. This fixes an issue that responses were not recognized anymore by the client after some time. - allow an observe request to be silently cancelled by the client, which is the suggested way per RFC 7641, in case a new observe request is started and the CLI user did not explicitly cancel the previous observe. This leaves the choice to the CLI user whether to explicitly cancel or just forget the request. - ensure that the client accepts CON notifications which are interspersed with NON notifications per RFC 7641. Previously, this caused the client to send RST instead of ACK. - avoids the error 28 ResponseTimeout popping up in various cases by keeping the observe request active. - implements the mandatory interspersing of CON notifications when a NON observe relation is ongoing, per RFC 7641. This is done by sending a CON notification after every 5 NON notifications, same as done by libcoap. When such CON notification times out (i.e. undelivered/unack'ed) then the observe subscription is automatically cleared after all its retries have been made. During this effort of trying to deliver the notification, the NON notifications (in case these follow) are still being sent in fire-and-forget mode as usual. - if already one subscription is ongoing, the server will ignore further subscription requests (Observe Option) per RFC 7641 Section 4.1 and treat the request normally. - log message at server side when a subscriber is cancelled. Fixes #11971
2026-06-06 05:24:51 +00:00 · 2025-11-18 19:35:54 +01:00
parent 249ce78cf9
commit 3dd2d471ea
4 changed files with 206 additions and 40 deletions
@@ -53,6 +53,8 @@ Coap::Coap(otInstance *aInstance, OutputImplementer &aOutputImplementer)
    , mRequestTokenLength(0)
    , mSubscriberTokenLength(0)
    , mSubscriberConfirmableNotifications(false)
+    , mValidateObserveClient(false)
+    , mNotificationSeriesCount(0)
 #endif
 #if OPENTHREAD_CONFIG_COAP_BLOCKWISE_TRANSFER_ENABLE
    , mBlockCount(1)
@@ -72,7 +74,7 @@ Coap::Coap(otInstance *aInstance, OutputImplementer &aOutputImplementer)
 }

 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
-otError Coap::CancelResourceSubscription(void)
+otError Coap::CancelResourceSubscription(bool aSendCancelMessage)
 {
    otError       error   = OT_ERROR_NONE;
    otMessage    *message = nullptr;
@@ -84,15 +86,18 @@ otError Coap::CancelResourceSubscription(void)

    VerifyOrExit(mRequestTokenLength != 0, error = OT_ERROR_INVALID_STATE);

-    message = otCoapNewMessage(GetInstancePtr(), nullptr);
-    VerifyOrExit(message != nullptr, error = OT_ERROR_NO_BUFS);
+    if (aSendCancelMessage)
+    {
+        message = otCoapNewMessage(GetInstancePtr(), nullptr);
+        VerifyOrExit(message != nullptr, error = OT_ERROR_NO_BUFS);

-    otCoapMessageInit(message, OT_COAP_TYPE_CONFIRMABLE, OT_COAP_CODE_GET);
+        otCoapMessageInit(message, OT_COAP_TYPE_CONFIRMABLE, OT_COAP_CODE_GET);

-    SuccessOrExit(error = otCoapMessageSetToken(message, mRequestToken, mRequestTokenLength));
-    SuccessOrExit(error = otCoapMessageAppendObserveOption(message, 1));
-    SuccessOrExit(error = otCoapMessageAppendUriPathOptions(message, mRequestUri));
-    SuccessOrExit(error = otCoapSendRequest(GetInstancePtr(), message, &messageInfo, &Coap::HandleResponse, this));
+        SuccessOrExit(error = otCoapMessageSetToken(message, mRequestToken, mRequestTokenLength));
+        SuccessOrExit(error = otCoapMessageAppendObserveOption(message, 1));
+        SuccessOrExit(error = otCoapMessageAppendUriPathOptions(message, mRequestUri));
+        SuccessOrExit(error = otCoapSendRequest(GetInstancePtr(), message, &messageInfo, &Coap::HandleResponse, this));
+    }

    ClearAllBytes(mRequestAddr);
    ClearAllBytes(mRequestUri);
@@ -110,6 +115,8 @@ exit:

 void Coap::CancelSubscriber(void)
 {
+    OutputFormat("Removed subscriber ");
+    OutputIp6AddressLine(mSubscriberSock.mAddress);
    ClearAllBytes(mSubscriberSock);
    mSubscriberTokenLength = 0;
 }
@@ -157,7 +164,7 @@ template <> otError Coap::Process<Cmd("cancel")>(Arg aArgs[])
 {
    OT_UNUSED_VARIABLE(aArgs);

-    return CancelResourceSubscription();
+    return CancelResourceSubscription(true);
 }
 #endif

@@ -248,6 +255,10 @@ template <> otError Coap::Process<Cmd("set")>(Arg aArgs[])
 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
        if (mSubscriberTokenLength > 0)
        {
+            // determine the type of notification to send
+            const bool isConNotification = mSubscriberConfirmableNotifications || mValidateObserveClient;
+            mValidateObserveClient       = false;
+
            // Notify the subscriber
            ClearAllBytes(messageInfo);
            messageInfo.mPeerAddr = mSubscriberSock.mAddress;
@@ -259,19 +270,40 @@ template <> otError Coap::Process<Cmd("set")>(Arg aArgs[])
            notificationMessage = otCoapNewMessage(GetInstancePtr(), nullptr);
            VerifyOrExit(notificationMessage != nullptr, error = OT_ERROR_NO_BUFS);

-            otCoapMessageInit(
-                notificationMessage,
-                ((mSubscriberConfirmableNotifications) ? OT_COAP_TYPE_CONFIRMABLE : OT_COAP_TYPE_NON_CONFIRMABLE),
-                OT_COAP_CODE_CONTENT);
+            otCoapMessageInit(notificationMessage,
+                              (isConNotification ? OT_COAP_TYPE_CONFIRMABLE : OT_COAP_TYPE_NON_CONFIRMABLE),
+                              OT_COAP_CODE_CONTENT);

            SuccessOrExit(error = otCoapMessageSetToken(notificationMessage, mSubscriberToken, mSubscriberTokenLength));
            SuccessOrExit(error = otCoapMessageAppendObserveOption(notificationMessage, mObserveSerial++));
            SuccessOrExit(error = otCoapMessageSetPayloadMarker(notificationMessage));
            SuccessOrExit(error = otMessageAppend(notificationMessage, mResourceContent,
                                                  static_cast<uint16_t>(strlen(mResourceContent))));
+            // Send the CoAP notification message, which is a CoAP Response, using the send-request API.
+            // If NON, it's fire-and-forget. If CON, supply the handler for receiving the CoAP ACK.
+            SuccessOrExit(
+                error = otCoapSendRequest(
+                    GetInstancePtr(), notificationMessage, &messageInfo,
+                    isConNotification ? static_cast<otCoapResponseHandler>(&Coap::HandleNotificationAck) : nullptr,
+                    this));

-            SuccessOrExit(error = otCoapSendRequest(GetInstancePtr(), notificationMessage, &messageInfo,
-                                                    &Coap::HandleNotificationResponse, this));
+            // after starting a single client validation, don't require validation for the
+            // kMaxNonNotificationsBeforeValidation next notifications. Note that if mSubscriberConfirmableNotifications
+            // == true, validation is done automatically with every CON notification.
+            if (isConNotification)
+            {
+                mNotificationSeriesCount = 0;
+            }
+            else
+            {
+                mNotificationSeriesCount++;
+                if (mNotificationSeriesCount >= kMaxNonNotificationsBeforeValidation)
+                {
+                    mNotificationSeriesCount = 0;
+                    // Require client validation using a CON message for the next notification.
+                    mValidateObserveClient = true;
+                }
+            }
        }
 #endif // OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
    }
@@ -665,8 +697,9 @@ otError Coap::ProcessRequest(Arg aArgs[], otCoapCode aCoapCode)
 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
    if (aCoapObserve && mRequestTokenLength)
    {
-        // New observe request, cancel any existing observation
-        SuccessOrExit(error = CancelResourceSubscription());
+        // New observe request: cancel the existing observation silently, without sending an explicit cancel message.
+        // Note: explicit cancellation MAY be sent by the user using 'coap cancel' per Section 3.6 of RFC7641.
+        IgnoreError(CancelResourceSubscription(false));
    }
 #endif

@@ -909,12 +942,12 @@ void Coap::HandleRequest(otMessage *aMessage, const otMessageInfo *aMessageInfo)
 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
        if (observePresent && (mSubscriberTokenLength > 0) && (observe == 0))
        {
-            // There is already a subscriber
-            responseCode = OT_COAP_CODE_SERVICE_UNAVAILABLE;
+            // There is already a subscriber: ignore the Observe option per Section 4.1 of RFC7641.
+            observePresent = false;
        }
-        else
 #endif
-            if (otCoapMessageGetCode(aMessage) == OT_COAP_CODE_GET)
+
+        if (otCoapMessageGetCode(aMessage) == OT_COAP_CODE_GET)
        {
            responseCode = OT_COAP_CODE_CONTENT;
 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
@@ -927,6 +960,8 @@ void Coap::HandleRequest(otMessage *aMessage, const otMessageInfo *aMessageInfo)
                    mSubscriberSock.mAddress = aMessageInfo->mPeerAddr;
                    mSubscriberSock.mPort    = aMessageInfo->mPeerPort;
                    mSubscriberTokenLength   = otCoapMessageGetTokenLength(aMessage);
+                    mValidateObserveClient   = false;
+                    mNotificationSeriesCount = 0;
                    memcpy(mSubscriberToken, otCoapMessageGetToken(aMessage), mSubscriberTokenLength);

                    /*
@@ -960,8 +995,11 @@ void Coap::HandleRequest(otMessage *aMessage, const otMessageInfo *aMessageInfo)
        responseMessage = otCoapNewMessage(GetInstancePtr(), nullptr);
        VerifyOrExit(responseMessage != nullptr, error = OT_ERROR_NO_BUFS);

-        SuccessOrExit(
-            error = otCoapMessageInitResponse(responseMessage, aMessage, OT_COAP_TYPE_ACKNOWLEDGMENT, responseCode));
+        SuccessOrExit(error = otCoapMessageInitResponse(responseMessage, aMessage,
+                                                        otCoapMessageGetType(aMessage) == OT_COAP_TYPE_CONFIRMABLE
+                                                            ? OT_COAP_TYPE_ACKNOWLEDGMENT
+                                                            : OT_COAP_TYPE_NON_CONFIRMABLE,
+                                                        responseCode));

        if (responseCode == OT_COAP_CODE_CONTENT)
        {
@@ -1024,15 +1062,12 @@ exit:
 }

 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
-void Coap::HandleNotificationResponse(void                *aContext,
-                                      otMessage           *aMessage,
-                                      const otMessageInfo *aMessageInfo,
-                                      otError              aError)
+void Coap::HandleNotificationAck(void *aContext, otMessage *aMessage, const otMessageInfo *aMessageInfo, otError aError)
 {
-    static_cast<Coap *>(aContext)->HandleNotificationResponse(aMessage, aMessageInfo, aError);
+    static_cast<Coap *>(aContext)->HandleNotificationAck(aMessage, aMessageInfo, aError);
 }

-void Coap::HandleNotificationResponse(otMessage *aMessage, const otMessageInfo *aMessageInfo, otError aError)
+void Coap::HandleNotificationAck(otMessage *aMessage, const otMessageInfo *aMessageInfo, otError aError)
 {
    OT_UNUSED_VARIABLE(aMessage);

@@ -1041,13 +1076,15 @@ void Coap::HandleNotificationResponse(otMessage *aMessage, const otMessageInfo *
    case OT_ERROR_NONE:
        if (aMessageInfo != nullptr)
        {
-            OutputFormat("Received ACK in reply to notification from ");
+            OutputFormat("Received coap notification ACK from ");
            OutputIp6AddressLine(aMessageInfo->mPeerAddr);
        }
        break;

    default:
-        OutputLine("coap receive notification response error %d: %s", aError, otThreadErrorToString(aError));
+        // A single CON notification delivery failure will cancel the subscription. RFC7641 would also allow
+        // retrying CON notification delivery more often - this is implementation-specific.
+        OutputLine("coap notification delivery error %d: %s", aError, otThreadErrorToString(aError));
        CancelSubscriber();
        break;
    }
@@ -89,7 +89,7 @@ private:
    template <CommandId kCommandId> otError Process(Arg aArgs[]);

 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
-    otError CancelResourceSubscription(void);
+    otError CancelResourceSubscription(bool aSendCancelMessage);
    void    CancelSubscriber(void);
 #endif

@@ -105,11 +105,16 @@ private:
    void        HandleRequest(otMessage *aMessage, const otMessageInfo *aMessageInfo);

 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
-    static void HandleNotificationResponse(void                *aContext,
-                                           otMessage           *aMessage,
-                                           const otMessageInfo *aMessageInfo,
-                                           otError              aError);
-    void        HandleNotificationResponse(otMessage *aMessage, const otMessageInfo *aMessageInfo, otError aError);
+    // Maximum number of non-confirmable (NON) notifications that are sent, before sending a confirmable notification
+    // for validating that the client is still there and still interested. The value (5) matches libcoap's
+    // implementation. The mechanism satisfies RFC 7641 requirements for interspersing confirmable (CON) notifications
+    // among non-confirmable (NON) notifications to validate the client.
+    static constexpr uint8_t kMaxNonNotificationsBeforeValidation = 5;
+    static void              HandleNotificationAck(void                *aContext,
+                                                   otMessage           *aMessage,
+                                                   const otMessageInfo *aMessageInfo,
+                                                   otError              aError);
+    void HandleNotificationAck(otMessage *aMessage, const otMessageInfo *aMessageInfo, otError aError);
 #endif

    static void HandleResponse(void *aContext, otMessage *aMessage, const otMessageInfo *aMessageInfo, otError aError);
@@ -171,6 +176,8 @@ private:
    uint8_t  mRequestTokenLength;
    uint8_t  mSubscriberTokenLength;
    bool     mSubscriberConfirmableNotifications;
+    bool     mValidateObserveClient;
+    uint8_t  mNotificationSeriesCount;
 #endif
 #if OPENTHREAD_CONFIG_COAP_BLOCKWISE_TRANSFER_ENABLE
    uint32_t mBlockCount;
@@ -1121,7 +1121,7 @@ void CoapBase::ProcessReceivedResponse(Message &aMessage, const Ip6::MessageInfo
            FinalizeCoapTransaction(*request, metadata, nullptr, nullptr, kErrorAbort);
        }

-        // Silently ignore non-empty reset messages (RFC 7252, p. 4.2).
+        // Silently ignore non-empty reset messages (RFC 7252, Section 4.2).
        break;

    case kTypeAck:
@@ -1131,7 +1131,7 @@ void CoapBase::ProcessReceivedResponse(Message &aMessage, const Ip6::MessageInfo
 #if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
            if (metadata.mObserve && !request->IsRequest())
            {
-                // This is the ACK to our RFC7641 notification.  There will be no
+                // This is the ACK to our RFC7641 CON notification.  There will be no
                // "separate" response so pass it back as if it were a piggy-backed
                // response so we can stop re-sending and the application can move on.
                FinalizeCoapTransaction(*request, metadata, &aMessage, &aMessageInfo, kErrorNone);
@@ -1278,8 +1278,17 @@ void CoapBase::ProcessReceivedResponse(Message &aMessage, const Ip6::MessageInfo
 #endif
                                                           ))
        {
-            // If multicast non-confirmable request, allow multiple responses
            metadata.mResponseHandler(metadata.mResponseContext, &aMessage, &aMessageInfo, kErrorNone);
+
+#if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
+            // When any Observe response is seen, consider a NON observe request "acknowledged" at this point.
+            // This will keep the Observe request active indefinitely until it is canceled.
+            if (metadata.mObserve && !metadata.mConfirmable && responseObserve)
+            {
+                metadata.mAcknowledged = true;
+                metadata.UpdateIn(*request);
+            }
+#endif
        }
        else
        {
@@ -0,0 +1,113 @@
+#!/usr/bin/expect -f
+#
+#  Copyright (c) 2025, The OpenThread Authors.
+#  All rights reserved.
+#
+#  Redistribution and use in source and binary forms, with or without
+#  modification, are permitted provided that the following conditions are met:
+#  1. Redistributions of source code must retain the above copyright
+#     notice, this list of conditions and the following disclaimer.
+#  2. Redistributions in binary form must reproduce the above copyright
+#     notice, this list of conditions and the following disclaimer in the
+#     documentation and/or other materials provided with the distribution.
+#  3. Neither the name of the copyright holder nor the
+#     names of its contributors may be used to endorse or promote products
+#     derived from this software without specific prior written permission.
+#
+#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+#  POSSIBILITY OF SUCH DAMAGE.
+#
+
+source "tests/scripts/expect/_common.exp"
+source "tests/scripts/expect/_multinode.exp"
+
+setup_two_nodes
+
+# Start CoAP server and set resource contents
+switch_node 1
+send "coap start\n"
+expect_line "Done"
+send "coap resource test/resource\n"
+expect_line "Done"
+send "coap set Testing123\n"
+expect_line "Done"
+set addr_1 [get_ipaddr mleid]
+
+# Start CoAP client and observe resource
+switch_node 2
+send "coap start\n"
+expect_line "Done"
+send "coap observe $addr_1 test/resource\n"
+expect_line "Done"
+expect "coap response from $addr_1 OBS=0 with payload: 54657374696e67313233" # ASCII of "Testing123"
+
+set addr_2 [get_ipaddr mleid]
+
+# Verify that server creates the subscription
+switch_node 1
+expect "coap request from $addr_2 GET OBS=0"
+expect "Subscribing client"
+expect "coap response sent"
+send "coap resource\n"
+expect "test/resource"
+expect_line "Done"
+send "coap set\n"
+expect "Testing123"
+expect_line "Done"
+
+# Server modifies resource several times; client gets notification. The 6th notification is sent as CON,
+# to validate the client's interest to keep receiving these notifications.
+for {set i 1} {$i <= 9} {incr i} {
+    switch_node 1
+    send "coap set TestValue2_$i\n"
+    expect "sending coap notification to $addr_2"
+    expect_line "Done"
+    switch_node 2
+    expect "coap response from $addr_1 OBS=$i with payload: 5465737456616c7565325f[binary encode hex $i]"
+    if {$i == 6} {
+        switch_node 1
+        expect "Received coap notification ACK from $addr_2"
+    }
+}
+
+# Client cancels subscription, server removes it.
+send "coap cancel\n"
+expect_line "Done"
+switch_node 1
+expect "Removed subscriber $addr_2"
+
+# CoAP client performs CON observe, server sends piggybacked response.
+switch_node 2
+send "coap observe $addr_1 test/resource con\n"
+expect_line "Done"
+expect "coap response from $addr_1 OBS=10 with payload: 5465737456616c7565325f39" # value from previous for loop
+switch_node 1
+
+# Server sends CON notification, client sends empty Ack.
+switch_node 1
+send "coap set TestValue3\n"
+expect "sending coap notification to $addr_2"
+expect_line "Done"
+expect "Received coap notification ACK from $addr_2"
+switch_node 2
+expect "coap response from $addr_1 OBS=11 with payload: 5465737456616c756533"
+
+# Server sends next CON notification, client sends empty Ack.
+switch_node 1
+send "coap set TestValue4\n"
+expect "sending coap notification to $addr_2"
+expect_line "Done"
+expect "Received coap notification ACK from $addr_2"
+switch_node 2
+expect "coap response from $addr_1 OBS=12 with payload: 5465737456616c756534"
+
+dispose_all