varlink: fix varlink_collect_full not resetting state

The varlink_collect_full function did not set varlink client's state
when the reply was an error. The state was stuck in "collecting-reply".

I discovered that while hacking on network varlink interface (adding a
new varlink method). The debug logs shows the process of performing the
first query which replies with an error:

  varlink: Setting state idle-client
  network: Sending message: {"method":"io.systemd.Network.LLDPNeighbors","parameters":{"ifindex":1},"more":true}
  network: Changing state idle-client → collecting
  network: Received message: {"error":"org.varlink.service.MethodNotFound","parameters":{"method":"io.systemd.Network.LLDPNeighbors"}}
  network: Changing state collecting → collecting-reply

Now another varlink_collect call is being made, but

  network: Connection busy.
  Failed to execute varlink call: Device or resource busy

This was not caught by the tests because there were no varlink_collect
calls that resulted in error reply.

src/shared/varlink.c

@@ -2390,6 +2390,13 @@ int varlink_collect_full(
                         JsonVariant *e = json_variant_by_key(v->current, "error"),
                                 *p = json_variant_by_key(v->current, "parameters");
 
+                        /* Unless there is more to collect we reset state to idle */
+                        if (!FLAGS_SET(v->current_reply_flags, VARLINK_REPLY_CONTINUES)) {
+                                varlink_set_state(v, VARLINK_IDLE_CLIENT);
+                                assert(v->n_pending == 1);
+                                v->n_pending--;
+                        }
+
                         if (e) {
                                 if (!ret_error_id)
                                         return varlink_error_to_errno(json_variant_string(e), p);
@@ -2418,10 +2425,6 @@ int varlink_collect_full(
                                 continue;
                         }
 
-                        varlink_set_state(v, VARLINK_IDLE_CLIENT);
-                        assert(v->n_pending == 1);
-                        v->n_pending--;
-
                         if (ret_parameters)
                                 /* Install the collection array in the connection object, so that we can hand
                                  * out a pointer to it without passing over ownership, to make it work more

src/test/test-varlink.c

@@ -239,6 +239,7 @@ static void flood_test(const char *address) {
 static void *thread(void *arg) {
         _cleanup_(varlink_flush_close_unrefp) Varlink *c = NULL;
         _cleanup_(json_variant_unrefp) JsonVariant *i = NULL;
+        _cleanup_(json_variant_unrefp) JsonVariant *wrong = NULL;
         JsonVariant *o = NULL, *k = NULL, *j = NULL;
         const char *error_id;
         const char *e;
@@ -252,6 +253,13 @@ static void *thread(void *arg) {
         assert_se(varlink_set_allow_fd_passing_input(c, true) >= 0);
         assert_se(varlink_set_allow_fd_passing_output(c, true) >= 0);
 
+        /* Test that client is able to perform two sequential varlink_collect calls if first resulted in an error */
+        assert_se(json_build(&wrong, JSON_BUILD_OBJECT(JSON_BUILD_PAIR("a", JSON_BUILD_INTEGER(88)),
+                                                       JSON_BUILD_PAIR("c", JSON_BUILD_INTEGER(99)))) >= 0);
+        assert_se(varlink_collect(c, "io.test.DoSomethingMore", wrong, &j, &error_id) >= 0);
+        assert_se(strcmp_ptr(error_id, "org.varlink.service.InvalidParameter") == 0);
+
+
         assert_se(varlink_collect(c, "io.test.DoSomethingMore", i, &j, &error_id) >= 0);
 
         assert_se(!error_id);