commit 888a79e

Call process_input_message() from the shared-inbox input.

This way, some garbage like unrequested Deletes from Mastodon
and other transient errors (like unaccessible authors) can be
short-circuited before propagating the message to the users.

 1@@ -2052,45 +2052,66 @@ void process_queue_item(xs_dict *q_item)
 2     }
 3     else
 4     if (strcmp(type, "input") == 0) {
 5-        /* redistribute the input message to all users */
 6-        char *ntid   = xs_dict_get(q_item, "ntid");
 7-        xs *tmpfn    = xs_fmt("%s/tmp/%s.json", srv_basedir, ntid);
 8         xs_dict *msg = xs_dict_get(q_item, "message");
 9-        FILE *f;
10+        xs_dict *req = xs_dict_get(q_item, "req");
11+        int retries  = xs_number_get(xs_dict_get(q_item, "retries"));
12+
13+        /* do some instance-level checks */
14+        int r = process_input_message(NULL, msg, req);
15 
16-        if ((f = fopen(tmpfn, "w")) != NULL) {
17-            xs_json_dump(q_item, 4, f);
18-            fclose(f);
19+        if (r == 0) {
20+            /* transient error? retry */
21+            int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
22+
23+            if (retries > queue_retry_max)
24+                srv_log(xs_fmt("shared input giving up"));
25+            else {
26+                /* reenqueue */
27+                enqueue_shared_input(msg, req, retries + 1);
28+                srv_log(xs_fmt("shared input requeue #%d", retries + 1));
29+            }
30         }
31+        else
32+        if (r == 2) {
33+            /* redistribute the input message to all users */
34+            char *ntid = xs_dict_get(q_item, "ntid");
35+            xs *tmpfn  = xs_fmt("%s/tmp/%s.json", srv_basedir, ntid);
36+            FILE *f;
37+
38+            if ((f = fopen(tmpfn, "w")) != NULL) {
39+                xs_json_dump(q_item, 4, f);
40+                fclose(f);
41+            }
42 
43-        xs *users = user_list();
44-        xs_list *p = users;
45-        char *v;
46-        int cnt = 0;
47+            xs *users = user_list();
48+            xs_list *p = users;
49+            char *v;
50+            int cnt = 0;
51 
52-        while (xs_list_iter(&p, &v)) {
53-            snac user;
54+            while (xs_list_iter(&p, &v)) {
55+                snac user;
56 
57-            if (user_open(&user, v)) {
58-                if (is_msg_for_me(&user, msg)) {
59-                    xs *fn = xs_fmt("%s/queue/%s.json", user.basedir, ntid);
60+                if (user_open(&user, v)) {
61+                    if (is_msg_for_me(&user, msg)) {
62+                        xs *fn = xs_fmt("%s/queue/%s.json", user.basedir, ntid);
63 
64-                    snac_debug(&user, 1, xs_fmt("enqueue_input (from shared inbox) %s", fn));
65+                        snac_debug(&user, 1, xs_fmt("enqueue_input (from shared inbox) %s", fn));
66 
67-                    if (link(tmpfn, fn) < 0)
68-                        srv_log(xs_fmt("link(%s, %s) error", tmpfn, fn));
69+                        if (link(tmpfn, fn) < 0)
70+                            srv_log(xs_fmt("link(%s, %s) error", tmpfn, fn));
71 
72-                    cnt++;
73-                }
74+                        cnt++;
75+                    }
76 
77-                user_free(&user);
78+                    user_free(&user);
79+                }
80             }
81-        }
82 
83-        unlink(tmpfn);
84+            unlink(tmpfn);
85 
86-        if (cnt == 0)
87-            srv_debug(1, xs_fmt("no valid recipients for %s", tmpfn));
88+            if (cnt == 0)
89+                srv_debug(1, xs_fmt("no valid recipients for %s", tmpfn));
90+        }
91     }
92     else
93         srv_log(xs_fmt("unexpected q_item type '%s'", type));