commit 60e8953
default
·
2023-09-29 08:34:22 +0000 UTC
parent 9cce2ee
Give more retry time to timedout connections.
3 files changed,
+21,
-11
+12,
-5
1@@ -1798,7 +1798,7 @@ void process_user_queue_item(snac *snac, xs_dict *q_item)
2 if (inbox != NULL) {
3 /* add to the set and, if it's not there, send message */
4 if (xs_set_add(&inboxes, inbox) == 1)
5- enqueue_output(snac, msg, inbox, 0);
6+ enqueue_output(snac, msg, inbox, 0, 0);
7 }
8 else
9 snac_log(snac, xs_fmt("cannot find inbox for %s", actor));
10@@ -1812,7 +1812,7 @@ void process_user_queue_item(snac *snac, xs_dict *q_item)
11 p = shibx;
12 while (xs_list_iter(&p, &inbox)) {
13 if (xs_set_add(&inboxes, inbox) == 1)
14- enqueue_output(snac, msg, inbox, 0);
15+ enqueue_output(snac, msg, inbox, 0, 0);
16 }
17 }
18
19@@ -1896,6 +1896,7 @@ void process_queue_item(xs_dict *q_item)
20 xs_str *seckey = xs_dict_get(q_item, "seckey");
21 xs_dict *msg = xs_dict_get(q_item, "message");
22 int retries = xs_number_get(xs_dict_get(q_item, "retries"));
23+ int p_status = xs_number_get(xs_dict_get(q_item, "p_status"));
24 xs *payload = NULL;
25 int p_size = 0;
26
27@@ -1909,8 +1910,9 @@ void process_queue_item(xs_dict *q_item)
28 return;
29 }
30
31- /* deliver */
32- status = send_to_inbox_raw(keyid, seckey, inbox, msg, &payload, &p_size, retries == 0 ? 3 : 8);
33+ /* deliver (if previous error status was a timeout, try now longer) */
34+ status = send_to_inbox_raw(keyid, seckey, inbox, msg,
35+ &payload, &p_size, p_status == 599 ? 20 : 3);
36
37 if (payload) {
38 if (p_size > 64) {
39@@ -1934,6 +1936,11 @@ void process_queue_item(xs_dict *q_item)
40 if (!valid_status(status)) {
41 retries++;
42
43+ /* if it's not the first time it fails with a timeout,
44+ penalize the server by skipping one retry */
45+ if (p_status == status && status == 499)
46+ retries++;
47+
48 /* error sending; requeue? */
49 if (status == 400 || status == 404 || status == 410 || status < 0)
50 /* explicit error: discard */
51@@ -1943,7 +1950,7 @@ void process_queue_item(xs_dict *q_item)
52 srv_log(xs_fmt("output message: giving up %s %d", inbox, status));
53 else {
54 /* requeue */
55- enqueue_output_raw(keyid, seckey, msg, inbox, retries);
56+ enqueue_output_raw(keyid, seckey, msg, inbox, retries, status);
57 srv_log(xs_fmt("output message: requeue %s #%d", inbox, retries));
58 }
59 }
M
data.c
+7,
-4
1@@ -2041,13 +2041,16 @@ void enqueue_input(snac *snac, const xs_dict *msg, const xs_dict *req, int retri
2
3
4 void enqueue_output_raw(const char *keyid, const char *seckey,
5- xs_dict *msg, xs_str *inbox, int retries)
6+ xs_dict *msg, xs_str *inbox, int retries, int p_status)
7 /* enqueues an output message to an inbox */
8 {
9 xs *qmsg = _new_qmsg("output", msg, retries);
10 char *ntid = xs_dict_get(qmsg, "ntid");
11 xs *fn = xs_fmt("%s/queue/%s.json", srv_basedir, ntid);
12
13+ xs *ns = xs_number_new(p_status);
14+ qmsg = xs_dict_append(qmsg, "p_status", ns);
15+
16 qmsg = xs_dict_append(qmsg, "inbox", inbox);
17 qmsg = xs_dict_append(qmsg, "keyid", keyid);
18 qmsg = xs_dict_append(qmsg, "seckey", seckey);
19@@ -2062,7 +2065,7 @@ void enqueue_output_raw(const char *keyid, const char *seckey,
20 }
21
22
23-void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries)
24+void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries, int p_status)
25 /* enqueues an output message to an inbox */
26 {
27 if (xs_startswith(inbox, snac->actor)) {
28@@ -2072,7 +2075,7 @@ void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries)
29
30 char *seckey = xs_dict_get(snac->key, "secret");
31
32- enqueue_output_raw(snac->actor, seckey, msg, inbox, retries);
33+ enqueue_output_raw(snac->actor, seckey, msg, inbox, retries, p_status);
34 }
35
36
37@@ -2082,7 +2085,7 @@ void enqueue_output_by_actor(snac *snac, xs_dict *msg, const xs_str *actor, int
38 xs *inbox = get_actor_inbox(snac, actor);
39
40 if (!xs_is_null(inbox))
41- enqueue_output(snac, msg, inbox, retries);
42+ enqueue_output(snac, msg, inbox, retries, 0);
43 else
44 snac_log(snac, xs_fmt("enqueue_output_by_actor cannot get inbox %s", actor));
45 }
M
snac.h
+2,
-2
1@@ -178,8 +178,8 @@ int instance_unblock(const char *instance);
2
3 void enqueue_input(snac *snac, const xs_dict *msg, const xs_dict *req, int retries);
4 void enqueue_output_raw(const char *keyid, const char *seckey,
5- xs_dict *msg, xs_str *inbox, int retries);
6-void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries);
7+ xs_dict *msg, xs_str *inbox, int retries, int p_status);
8+void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries, int p_status);
9 void enqueue_output_by_actor(snac *snac, xs_dict *msg, const xs_str *actor, int retries);
10 void enqueue_email(xs_str *msg, int retries);
11 void enqueue_telegram(const xs_str *msg, const char *bot, const char *chat_id);