Block1: Handle large FETCH with large blocked request and Observe

Handle Observe cancellation with large blocked FETCH that needs to use
diffferent tokens when there is packet loss by checking the cache-key
if the token is not found (Q-Block1).

Let FETCH with Q-Block1 use NON now that cache-key is getting checked
if token mismatch.

Handle getting an unsolicited Observe response in the middle of
sending the FETCH blocks that ultimately will be doing the Observe
cancellation following re-assembly. (The unsolicited Observe token will
be the same token that is used to respond to the observe cancellation).

include/coap3/coap_subscribe_internal.h

@@ -105,7 +105,8 @@ coap_subscription_t *coap_add_observer(coap_resource_t *resource,
  * @param resource The observed resource.
  * @param session The observer's session
  * @param token The token that identifies this subscription or @c NULL for
- * any token.
+ * the first subscription.
+ *
  * @return A valid subscription if exists or @c NULL otherwise.
  */
 coap_subscription_t *coap_find_observer(coap_resource_t *resource,
@@ -125,20 +126,40 @@ void coap_touch_observer(coap_context_t *context,
  const coap_bin_const_t *token);
 
 /**
- * Removes any subscription for @p observer from @p resource and releases the
+ * Removes any subscription for @p session observer from @p resource and releases the
  * allocated storage. The result is @c 1 if an observation relationship with @p
- * observer and @p token existed, @c 0 otherwise.
+ * session observer and @p token existed, @c 0 otherwise.
  *
  * @param resource The observed resource.
  * @param session The observer's session.
  * @param token The token that identifies this subscription or @c NULL for
- * any token.
+ * the first subscription.
+ *
  * @return @c 1 if the observer has been deleted, @c 0 otherwise.
  */
 int coap_delete_observer(coap_resource_t *resource,
  coap_session_t *session,
  const coap_bin_const_t *token);
 
+/**
+ * Removes any subscription for @p session observer from @p resource and releases the
+ * allocated storage. The result is @c 1 if an observation relationship with @p
+ * session observer and @p token existed, or cache-key derived from @p request matches,
+ * @c 0 otherwise.
+ *
+ * @param resource The observed resource.
+ * @param session The observer's session.
+ * @param token The token that identifies this subscription or @c NULL for
+ * the first subscription.
+ * @param request The requesting PDU.
+ *
+ * @return @c 1 if the observer has been deleted, @c 0 otherwise.
+ */
+int coap_delete_observer_request(coap_resource_t *resource,
+ coap_session_t *session,
+ const coap_bin_const_t *token,
+ coap_pdu_t *request);
+
 /**
  * Removes any subscription for @p session and releases the allocated storage.
  *

src/coap_block.c

@@ -494,18 +494,8 @@ coap_cancel_observe_lkd(coap_session_t *session, coap_binary_t *token,
  lg_crcv->observe_set = 0;
  if (pdu == NULL)
  return 0;
-#if COAP_Q_BLOCK_SUPPORT
- if (pdu->code == COAP_REQUEST_CODE_FETCH && using_q_block1) {
- /* Have to make sure all gets through in case of packet loss */
- pdu->type = COAP_MESSAGE_CON;
- } else {
- /* Need to make sure that this is the correct requested type */
- pdu->type = type;
- }
-#else /* ! COAP_Q_BLOCK_SUPPORT */
  /* Need to make sure that this is the correct requested type */
  pdu->type = type;
-#endif /* ! COAP_Q_BLOCK_SUPPORT */
 
  coap_update_option(pdu, COAP_OPTION_OBSERVE,
  coap_encode_var_safe(buf, sizeof(buf),
@@ -2080,13 +2070,19 @@ track_fetch_observe(coap_pdu_t *pdu, coap_lg_crcv_t *lg_crcv,
  coap_opt_length(opt));
  if (observe_action == COAP_OBSERVE_ESTABLISH) {
  /* Save the token in lg_crcv */
- tmp = coap_realloc_type(COAP_STRING, lg_crcv->obs_token,
- (block_num + 1) * sizeof(lg_crcv->obs_token[0]));
- if (tmp == NULL)
- return NULL;
- lg_crcv->obs_token = tmp;
- if (block_num + 1 == lg_crcv->obs_token_cnt)
- coap_delete_bin_const(lg_crcv->obs_token[block_num]);
+ if (lg_crcv->obs_token_cnt <= block_num) {
+ size_t i;
+
+ tmp = coap_realloc_type(COAP_STRING, lg_crcv->obs_token,
+ (block_num + 1) * sizeof(lg_crcv->obs_token[0]));
+ if (tmp == NULL)
+ return NULL;
+ lg_crcv->obs_token = tmp;
+ for (i = lg_crcv->obs_token_cnt; i < block_num + 1; i++) {
+ lg_crcv->obs_token[i] = NULL;
+ }
+ }
+ coap_delete_bin_const(lg_crcv->obs_token[block_num]);
 
  lg_crcv->obs_token_cnt = block_num + 1;
  lg_crcv->obs_token[block_num] = coap_new_bin_const(token->s,
@@ -2096,9 +2092,7 @@ track_fetch_observe(coap_pdu_t *pdu, coap_lg_crcv_t *lg_crcv,
  } else if (observe_action == COAP_OBSERVE_CANCEL) {
  /* Use the token in lg_crcv */
  if (block_num < lg_crcv->obs_token_cnt) {
- if (lg_crcv->obs_token[block_num]) {
- return lg_crcv->obs_token[block_num];
- }
+ return lg_crcv->obs_token[block_num];
  }
  }
  }
@@ -3457,6 +3451,36 @@ coap_handle_response_send_block(coap_session_t *session, coap_pdu_t *sent,
 #endif /* ! COAP_Q_BLOCK_SUPPORT */
  return 1;
  }
+ } else if (COAP_RESPONSE_CLASS(rcvd->code) == 2) {
+ /*
+ * Not a block response asking for the next block.
+ * Could be an Observe response overlapping with block FETCH doing
+ * Observe cancellation.
+ */
+ coap_opt_iterator_t opt_iter;
+ coap_opt_t *obs_opt;
+ int observe_action = -1;
+
+ if (p->pdu.code != COAP_REQUEST_CODE_FETCH) {
+ goto lg_xmit_finished;
+ }
+ obs_opt = coap_check_option(&p->pdu,
+ COAP_OPTION_OBSERVE,
+ &opt_iter);
+ if (obs_opt) {
+ observe_action = coap_decode_var_bytes(coap_opt_value(obs_opt),
+ coap_opt_length(obs_opt));
+ }
+ if (observe_action != COAP_OBSERVE_CANCEL) {
+ goto lg_xmit_finished;
+ }
+ obs_opt = coap_check_option(rcvd,
+ COAP_OPTION_OBSERVE,
+ &opt_iter);
+ if (obs_opt) {
+ return 0;
+ }
+ goto lg_xmit_finished;
  } else if (rcvd->code == COAP_RESPONSE_CODE(401)) {
  if (check_freshness(session, rcvd, sent, p, NULL))
  return 1;

src/coap_net.c

@@ -1502,25 +1502,6 @@ coap_send_lkd(coap_session_t *session, coap_pdu_t *pdu) {
  buf);
  }
  }
- if (pdu->type == COAP_MESSAGE_NON && pdu->code == COAP_REQUEST_CODE_FETCH &&
- coap_check_option(pdu, COAP_OPTION_OBSERVE, &opt_iter) &&
- coap_check_option(pdu, COAP_OPTION_Q_BLOCK1, &opt_iter)) {
- /* Issue with Fetch + Observe + Q-Block1 + NON if there are
- * retransmits as potential for Token confusion */
- pdu->type = COAP_MESSAGE_CON;
- /* Need to update associated lg_xmit */
- coap_lg_xmit_t *lg_xmit;
-
- LL_FOREACH(session->lg_xmit, lg_xmit) {
- if (lg_xmit->pdu.code == COAP_REQUEST_CODE_FETCH &&
- lg_xmit->b.b1.app_token &&
- coap_binary_equal(&pdu->actual_token, lg_xmit->b.b1.app_token)) {
- /* Update as this is a Request */
- lg_xmit->pdu.type = COAP_MESSAGE_CON;
- break;
- }
- }
- }
 #endif /* COAP_Q_BLOCK_SUPPORT */
 
  /*
@@ -3390,7 +3371,7 @@ handle_request(coap_context_t *context, coap_session_t *session, coap_pdu_t *pdu
  buf);
  }
  } else if (observe_action == COAP_OBSERVE_CANCEL) {
- coap_delete_observer(resource, session, &pdu->actual_token);
+ coap_delete_observer_request(resource, session, &pdu->actual_token, pdu);
  } else {
  coap_log_info("observe: unexpected action %d\n", observe_action);
  }

src/coap_resource.c

@@ -757,6 +757,10 @@ coap_find_observer_cache_key(coap_resource_t *resource, coap_session_t *session,
  return NULL;
 }
 
+/* https://rfc-editor.org/rfc/rfc7641#section-3.6 */
+static const uint16_t cache_ignore_options[] = { COAP_OPTION_ETAG,
+ COAP_OPTION_OSCORE
+ };
 coap_subscription_t *
 coap_add_observer(coap_resource_t *resource,
  coap_session_t *session,
@@ -766,10 +770,6 @@ coap_add_observer(coap_resource_t *resource,
  coap_cache_key_t *cache_key = NULL;
  size_t len;
  const uint8_t *data;
- /* https://rfc-editor.org/rfc/rfc7641#section-3.6 */
- static const uint16_t cache_ignore_options[] = { COAP_OPTION_ETAG,
- COAP_OPTION_OSCORE
- };
 
  assert(session);
 
@@ -964,14 +964,13 @@ coap_touch_observer(coap_context_t *context, coap_session_t *session,
  }
 }
 
-int
-coap_delete_observer(coap_resource_t *resource, coap_session_t *session,
- const coap_bin_const_t *token) {
- coap_subscription_t *s;
-
- s = coap_find_observer(resource, session, token);
+static void
+coap_delete_observer_internal(coap_resource_t *resource, coap_session_t *session,
+ coap_subscription_t *s) {
+ if (!s)
+ return;
 
- if (s && coap_get_log_level() >= COAP_LOG_DEBUG) {
+ if (coap_get_log_level() >= COAP_LOG_DEBUG) {
  char outbuf[2 * 8 + 1] = "";
  unsigned int i;
 
@@ -985,21 +984,67 @@ coap_delete_observer(coap_resource_t *resource, coap_session_t *session,
  (void *)s, outbuf, s->cache_key->key[0], s->cache_key->key[1],
  s->cache_key->key[2], s-> cache_key->key[3]);
  }
- if (s && session->context->observe_deleted)
+ if (session->context->observe_deleted)
  session->context->observe_deleted(session, s,
  session->context->observe_user_data);
 
- if (resource->subscribers && s) {
+ if (resource->subscribers) {
  LL_DELETE(resource->subscribers, s);
  coap_session_release_lkd(session);
  coap_delete_pdu(s->pdu);
  coap_delete_cache_key(s->cache_key);
  coap_free_type(COAP_SUBSCRIPTION, s);
  }
 
+ return;
+}
+
+int
+coap_delete_observer(coap_resource_t *resource, coap_session_t *session,
+ const coap_bin_const_t *token) {
+ coap_subscription_t *s;
+
+ s = coap_find_observer(resource, session, token);
+ if (s)
+ coap_delete_observer_internal(resource, session, s);
+
  return s != NULL;
 }
 
+int
+coap_delete_observer_request(coap_resource_t *resource, coap_session_t *session,
+ const coap_bin_const_t *token, coap_pdu_t *request) {
+ coap_subscription_t *s;
+ int ret = 0;
+
+ s = coap_find_observer(resource, session, token);
+ s = NULL;
+ if (!s) {
+ /*
+ * It is possible that the client is using the wrong token.
+ * An example being a large FETCH spanning multiple blocks.
+ */
+ coap_cache_key_t *cache_key;
+
+ cache_key = coap_cache_derive_key_w_ignore(session, request,
+ COAP_CACHE_IS_SESSION_BASED,
+ cache_ignore_options,
+ sizeof(cache_ignore_options)/sizeof(cache_ignore_options[0]));
+ if (cache_key) {
+ s = coap_find_observer_cache_key(resource, session, cache_key);
+ if (s) {
+ /* Delete entry with setup token */
+ ret = coap_delete_observer(resource, session, &s->pdu->actual_token);
+ }
+ coap_delete_cache_key(cache_key);
+ }
+ } else {
+ coap_delete_observer_internal(resource, session, s);
+ ret = 1;
+ }
+ return ret;
+}
+
 void
 coap_delete_observers(coap_context_t *context, coap_session_t *session) {
  RESOURCES_ITER(context->resources, resource) {