sbus: terminate ongoing chained requests if backend is restarted

If there is an outgoing request already chained and backend is
restarted, a new outgoing request is chained but not processed,
it waits for a timeout.

This patch makes sure that all outgoing requests are gracefully
terminated if backend restarts.

Resolves: #7503

src/monitor/monitor.c

@@ -1613,7 +1613,9 @@ static void monitor_sbus_connected(struct tevent_req *req)
     SBUS_INTERFACE(iface_service,
         sssd_service,
         SBUS_METHODS(SBUS_NO_METHODS),
-        SBUS_SIGNALS(SBUS_NO_SIGNALS),
+        SBUS_SIGNALS(
+            SBUS_EMITS(sssd_service, TerminateChainedRequests)
+        ),
         SBUS_PROPERTIES(
             SBUS_SYNC(GETTER, sssd_service, debug_level, generic_get_debug_level, NULL),
             SBUS_SYNC(SETTER, sssd_service, debug_level, generic_set_debug_level, NULL)
@@ -1810,6 +1812,9 @@ static void mt_svc_restart(struct tevent_context *ev,
     if (svc->type == MT_SVC_SERVICE) {
         add_new_service(svc->mt_ctx, svc->name, svc->restarts + 1);
     } else if (svc->type == MT_SVC_PROVIDER) {
+        sbus_emit_service_TerminateChainedRequests(svc->mt_ctx->sbus_conn,
+                                                   SSS_BUS_PATH,
+                                                   svc->busname);
         add_new_provider(svc->mt_ctx, svc->name, svc->restarts + 1);
     } else {
         /* Invalid type? */

src/responder/common/responder_iface.c

@@ -95,6 +95,19 @@ sss_resp_reset_ncache_groups(TALLOC_CTX *mem_ctx,
     return EOK;
 }
 
+static errno_t
+sss_resp_terminate_chained_requests(TALLOC_CTX *mem_ctx,
+                                    struct sbus_request *sbus_req,
+                                    struct resp_ctx *rctx,
+                                    const char *member)
+{
+    DEBUG(SSSDBG_TRACE_FUNC, "Teminating outgoing chained requests for: %s\n",
+          member);
+    sbus_connection_terminate_outgoing_member(sbus_req->conn, member);
+
+    return EOK;
+}
+
 errno_t
 sss_resp_register_sbus_iface(struct sbus_connection *conn,
                              struct resp_ctx *rctx)
@@ -111,7 +124,9 @@ sss_resp_register_sbus_iface(struct sbus_connection *conn,
         SBUS_LISTEN_SYNC(sssd_Responder_NegativeCache, ResetUsers,
                          SSS_BUS_PATH, sss_resp_reset_ncache_users, rctx),
         SBUS_LISTEN_SYNC(sssd_Responder_NegativeCache, ResetGroups,
-                         SSS_BUS_PATH, sss_resp_reset_ncache_groups, rctx)
+                         SSS_BUS_PATH, sss_resp_reset_ncache_groups, rctx),
+        SBUS_LISTEN_SYNC(sssd_service, TerminateChainedRequests,
+                         SSS_BUS_PATH, sss_resp_terminate_chained_requests, rctx)
     );
 
     ret = sbus_router_listen_map(conn, listeners);

src/sbus/connection/sbus_connection.c

@@ -471,3 +471,11 @@ void sbus_connection_free(struct sbus_connection *conn)
               conn);
     }
 }
+
+void
+sbus_connection_terminate_outgoing_member(struct sbus_connection *conn,
+                                          const char *member)
+{
+    sbus_requests_terminate_member(conn->requests->outgoing, member,
+                                   ERR_TERMINATED);
+}

src/sbus/request/sbus_request.c

@@ -449,6 +449,7 @@ static void sbus_incoming_request_sender_done(struct tevent_req *subreq)
     DBusMessageIter *write_iter = NULL;
     struct sbus_sender *sender;
     struct tevent_req *req;
+    const char *member = NULL;
     bool key_exists;
     errno_t ret;
 
@@ -463,6 +464,9 @@ static void sbus_incoming_request_sender_done(struct tevent_req *subreq)
     }
 
     state->request->sender = talloc_steal(state->request, sender);
+    if (sender != NULL) {
+        member = sender->name;
+    }
 
     ret = sbus_check_access(state->conn, state->request);
     if (ret != EOK) {
@@ -500,7 +504,7 @@ static void sbus_incoming_request_sender_done(struct tevent_req *subreq)
      * set a tevent callback that is triggered when the method handler is done.
      */
     ret = sbus_requests_add(state->conn->requests->incoming, state->key,
-                            state->conn, req, true, &key_exists);
+                            state->conn, req, member, true, &key_exists);
     if (ret != EOK || key_exists) {
         /* Cancel the sub request. Since there was either an error or the
          * sub request was chained. */
@@ -617,6 +621,7 @@ sbus_outgoing_request_send(TALLOC_CTX *mem_ctx,
     struct sbus_outgoing_request_state *state;
     struct tevent_req *subreq;
     struct tevent_req *req;
+    const char *destination;
     bool key_exists;
     errno_t ret;
 
@@ -645,6 +650,8 @@ sbus_outgoing_request_send(TALLOC_CTX *mem_ctx,
         }
     }
 
+    destination = dbus_message_get_destination(msg);
+
     /**
      * We will search table to see if the same request is not already
      * in progress. If it is, we register ourselves for notification
@@ -654,7 +661,7 @@ sbus_outgoing_request_send(TALLOC_CTX *mem_ctx,
      * set a tevent callback that is triggered when the method handler is done.
      */
     ret = sbus_requests_add(conn->requests->outgoing, key,
-                            conn, req, true, &key_exists);
+                            conn, req, destination, true, &key_exists);
     if (ret != EOK) {
         goto done;
     }
@@ -776,7 +783,7 @@ sbus_request_await_send(TALLOC_CTX *mem_ctx,
 
     /* Otherwise attach to this request. */
     ret = sbus_requests_add(conn->requests->outgoing, key, conn,
-                            req, false, NULL);
+                            req, member, false, NULL);
     if (ret != EOK) {
         DEBUG(SSSDBG_CRIT_FAILURE, "Unable to attach to the request list "
               "[%d]: %s\n", ret, sss_strerror(ret));

src/sbus/request/sbus_request_hash.c

@@ -118,6 +118,7 @@ sbus_requests_add(hash_table_t *table,
                   const char *key,
                   struct sbus_connection *conn,
                   struct tevent_req *req,
+                  const char *member,
                   bool is_dbus,
                   bool *_key_exists)
 {
@@ -150,6 +151,11 @@ sbus_requests_add(hash_table_t *table,
     item->req = req;
     item->conn = conn;
     item->is_dbus = is_dbus;
+    item->member = talloc_strdup(item, member);
+    if (member != NULL && item->member == NULL) {
+        ret = ENOMEM;
+        goto done;
+    }
 
     ret = sbus_requests_attach_spies(item);
     if (ret != EOK) {
@@ -323,3 +329,37 @@ sbus_requests_terminate_all(hash_table_t *table,
 
     talloc_free(values);
 }
+
+void
+sbus_requests_terminate_member(hash_table_t *table,
+                               const char *member,
+                               errno_t error)
+{
+    struct sbus_request_list *list;
+    struct sbus_request_list *item;
+    hash_value_t *values;
+    unsigned long int num;
+    unsigned long int i;
+    int hret;
+
+    hret = hash_values(table, &num, &values);
+    if (hret != HASH_SUCCESS) {
+        DEBUG(SSSDBG_CRIT_FAILURE, "Unable to get list of active requests "
+              "[%d]: %s\n", hret, hash_error_string(hret));
+        return;
+    }
+
+    for (i = 0; i < num; i++) {
+        list = sss_ptr_get_value(&values[i], struct sbus_request_list);
+        if ((member == NULL && list->member == NULL)
+            || (member != NULL && list->member != NULL && strcmp(member, list->member) == 0)) {
+            DLIST_FOR_EACH(item, list) {
+                sbus_requests_finish(item, error);
+            }
+        }
+
+        sbus_requests_delete(list);
+    }
+
+    talloc_free(values);
+}

src/sbus/sbus.h

@@ -336,6 +336,16 @@ errno_t
 sbus_connection_add_path_map(struct sbus_connection *conn,
                              struct sbus_path *map);
 
+/**
+ * Terminate all outgoing requests for given member.
+ *
+ * @param conn      An sbus connection.
+ * @param member D-Bus member name (destination)
+ */
+void
+sbus_connection_terminate_outgoing_member(struct sbus_connection *conn,
+                                          const char *member);
+
 /**
  * Add new signal listener to the router.
  *

src/sbus/sbus_private.h

@@ -431,6 +431,9 @@ struct sbus_request_list {
     struct tevent_req *req;
     struct sbus_connection *conn;
 
+    /* Member part of the key. Destination for outgoing, sender for incoming.*/
+    const char *member;
+
     bool is_invalid;
     bool is_dbus;
 
@@ -462,6 +465,7 @@ sbus_requests_add(hash_table_t *table,
                   const char *key,
                   struct sbus_connection *conn,
                   struct tevent_req *req,
+                  const char *member,
                   bool is_dbus,
                   bool *_key_exists);
 
@@ -484,6 +488,12 @@ void
 sbus_requests_terminate_all(hash_table_t *table,
                             errno_t error);
 
+/* Terminate requests associated with given member. */
+void
+sbus_requests_terminate_member(hash_table_t *table,
+                               const char *member,
+                               errno_t error);
+
 /* Create new sbus request. */
 struct sbus_request *
 sbus_request_create(TALLOC_CTX *mem_ctx,

src/sss_iface/sbus_sss_client_async.c

@@ -2376,6 +2376,16 @@ sbus_emit_resp_negcache_ResetUsers
         "sssd.Responder.NegativeCache", "ResetUsers");
 }
 
+void
+sbus_emit_monitor_SetActive
+    (struct sbus_connection *conn,
+     const char *object_path,
+     const char * arg_name)
+{
+    sbus_emit_signal_s(conn, object_path,
+        "sssd.monitor", "SetActive", arg_name);
+}
+
 void
 sbus_emit_nss_memcache_InvalidateAllGroups
     (struct sbus_connection *conn,
@@ -2412,3 +2422,13 @@ sbus_emit_nss_memcache_InvalidateGroupById
     sbus_emit_signal_u(conn, object_path,
         "sssd.nss.MemoryCache", "InvalidateGroupById", arg_gid);
 }
+
+void
+sbus_emit_service_TerminateChainedRequests
+    (struct sbus_connection *conn,
+     const char *object_path,
+     const char * arg_name)
+{
+    sbus_emit_signal_s(conn, object_path,
+        "sssd.service", "TerminateChainedRequests", arg_name);
+}

src/sss_iface/sbus_sss_client_async.h

@@ -428,6 +428,12 @@ sbus_emit_resp_negcache_ResetUsers
     (struct sbus_connection *conn,
      const char *object_path);
 
+void
+sbus_emit_monitor_SetActive
+    (struct sbus_connection *conn,
+     const char *object_path,
+     const char * arg_name);
+
 void
 sbus_emit_nss_memcache_InvalidateAllGroups
     (struct sbus_connection *conn,
@@ -449,4 +455,10 @@ sbus_emit_nss_memcache_InvalidateGroupById
      const char *object_path,
      uint32_t arg_gid);
 
+void
+sbus_emit_service_TerminateChainedRequests
+    (struct sbus_connection *conn,
+     const char *object_path,
+     const char * arg_name);
+
 #endif /* _SBUS_SSS_CLIENT_ASYNC_H_ */

src/sss_iface/sbus_sss_interface.h

@@ -677,6 +677,30 @@
         (handler_send), (handler_recv), (data)); \
 })
 
+/* Signal: sssd.monitor.SetActive */
+#define SBUS_SIGNAL_EMITS_sssd_monitor_SetActive() ({ \
+    sbus_signal("SetActive", \
+        _sbus_sss_args_sssd_monitor_SetActive, \
+        NULL); \
+})
+
+#define SBUS_SIGNAL_SYNC_sssd_monitor_SetActive(path, handler, data) ({ \
+    SBUS_CHECK_SYNC((handler), (data), const char *); \
+    sbus_listener_sync("sssd.monitor", "SetActive", (path), \
+        _sbus_sss_invoke_in_s_out__send, \
+        NULL, \
+        (handler), (data)); \
+})
+
+#define SBUS_SIGNAL_ASYNC_sssd_monitor_SetActive(path, handler_send, handler_recv, data) ({ \
+    SBUS_CHECK_SEND((handler_send), (data), const char *); \
+    SBUS_CHECK_RECV((handler_recv)); \
+    sbus_listener_async("sssd.monitor", "SetActive", (path), \
+        _sbus_sss_invoke_in_s_out__send, \
+        NULL, \
+        (handler_send), (handler_recv), (data)); \
+})
+
 /* Interface: sssd.nss.MemoryCache */
 #define SBUS_IFACE_sssd_nss_MemoryCache(methods, signals, properties) ({ \
     sbus_interface("sssd.nss.MemoryCache", NULL, \
@@ -961,6 +985,30 @@
         (handler_send), (handler_recv), (data)); \
 })
 
+/* Signal: sssd.service.TerminateChainedRequests */
+#define SBUS_SIGNAL_EMITS_sssd_service_TerminateChainedRequests() ({ \
+    sbus_signal("TerminateChainedRequests", \
+        _sbus_sss_args_sssd_service_TerminateChainedRequests, \
+        NULL); \
+})
+
+#define SBUS_SIGNAL_SYNC_sssd_service_TerminateChainedRequests(path, handler, data) ({ \
+    SBUS_CHECK_SYNC((handler), (data), const char *); \
+    sbus_listener_sync("sssd.service", "TerminateChainedRequests", (path), \
+        _sbus_sss_invoke_in_s_out__send, \
+        NULL, \
+        (handler), (data)); \
+})
+
+#define SBUS_SIGNAL_ASYNC_sssd_service_TerminateChainedRequests(path, handler_send, handler_recv, data) ({ \
+    SBUS_CHECK_SEND((handler_send), (data), const char *); \
+    SBUS_CHECK_RECV((handler_recv)); \
+    sbus_listener_async("sssd.service", "TerminateChainedRequests", (path), \
+        _sbus_sss_invoke_in_s_out__send, \
+        NULL, \
+        (handler_send), (handler_recv), (data)); \
+})
+
 /* Property: sssd.service.debug_level */
 #define SBUS_GETTER_SYNC_sssd_service_debug_level(handler, data) ({ \
     SBUS_CHECK_SYNC((handler), (data), uint32_t*); \

src/sss_iface/sbus_sss_symbols.c

@@ -339,6 +339,12 @@ _sbus_sss_args_sssd_monitor_RegisterService = {
     }
 };
 
+const struct sbus_argument
+_sbus_sss_args_sssd_monitor_SetActive[] = {
+    {.type = "s", .name = "name"},
+    {NULL}
+};
+
 const struct sbus_method_arguments
 _sbus_sss_args_sssd_nss_MemoryCache_UpdateInitgroups = {
     .input = (const struct sbus_argument[]){
@@ -442,3 +448,9 @@ _sbus_sss_args_sssd_service_sysbusReconnect = {
         {NULL}
     }
 };
+
+const struct sbus_argument
+_sbus_sss_args_sssd_service_TerminateChainedRequests[] = {
+    {.type = "s", .name = "name"},
+    {NULL}
+};

src/sss_iface/sbus_sss_symbols.h

@@ -100,6 +100,9 @@ _sbus_sss_args_sssd_dataprovider_sudoHandler;
 extern const struct sbus_method_arguments
 _sbus_sss_args_sssd_monitor_RegisterService;
 
+extern const struct sbus_argument
+_sbus_sss_args_sssd_monitor_SetActive[];
+
 extern const struct sbus_method_arguments
 _sbus_sss_args_sssd_nss_MemoryCache_UpdateInitgroups;
 
@@ -136,4 +139,7 @@ _sbus_sss_args_sssd_service_rotateLogs;
 extern const struct sbus_method_arguments
 _sbus_sss_args_sssd_service_sysbusReconnect;
 
+extern const struct sbus_argument
+_sbus_sss_args_sssd_service_TerminateChainedRequests[];
+
 #endif /* _SBUS_SSS_SYMBOLS_H_ */