The input queue also has retries.

3 files changed, 24 insertions, 12 deletions

diff --git a/activitypub.c b/activitypub.c
index 20c734b..8127e8d 100644
--- a/activitypub.c
+++ b/activitypub.c
@@ -557,7 +557,7 @@ d_char *msg_note(snac *snac, char *content, char *rcpts, char *in_reply_to)
 
 /** queues **/
 
-void process_message(snac *snac, char *msg, char *req)
+int process_message(snac *snac, char *msg, char *req)
 /* processes an ActivityPub message from the input queue */
 {
     /* actor and type exist, were checked previously */
@@ -575,10 +575,8 @@ void process_message(snac *snac, char *msg, char *req)
 
     /* bring the actor */
     if (!valid_status(actor_request(snac, actor, &actor_o))) {
-        /* error: re-enqueue to try later */
-        enqueue_input(snac, msg, req);
         snac_log(snac, xs_fmt("error requesting actor %s -- retry later", actor));
-        return;
+        return 0;
     }
 
     /* check the signature */
@@ -690,6 +688,8 @@ void process_message(snac *snac, char *msg, char *req)
 */
     else
         snac_debug(snac, 1, xs_fmt("process_message type '%s' ignored", type));
+
+    return 1;
 }
 
 
@@ -742,8 +742,17 @@ void process_queue(snac *snac)
             /* process the message */
             char *msg = xs_dict_get(q_item, "object");
             char *req = xs_dict_get(q_item, "req");
+            int retries = xs_number_get(xs_dict_get(q_item, "retries"));
 
-            process_message(snac, msg, req);
+            if (!process_message(snac, msg, req)) {
+                if (retries > queue_retry_max)
+                    snac_log(snac, xs_fmt("process_queue input giving up"));
+                else {
+                    /* reenqueue */
+                    enqueue_input(snac, msg, req, retries + 1);
+                    snac_log(snac, xs_fmt("process_queue input requeue %d", retries + 1));
+                }
+            }
         }
     }
 }
@@ -907,7 +916,7 @@ int activitypub_post_handler(d_char *req, char *q_path,
     }
 
     if (valid_status(status)) {
-        enqueue_input(&snac, msg, req);
+        enqueue_input(&snac, msg, req, 0);
         *ctype = "application/activity+json";
     }
 
diff --git a/data.c b/data.c
index c4ade59..ee21a98 100644
--- a/data.c
+++ b/data.c
@@ -831,21 +831,24 @@ int static_get(snac *snac, char *id, d_char **data, int *size)
 }
 
 
-void enqueue_input(snac *snac, char *msg, char *req)
+void enqueue_input(snac *snac, char *msg, char *req, int retries)
 /* enqueues an input message */
 {
-    xs *ntid = tid(0);
+    int qrt  = xs_number_get(xs_dict_get(srv_config, "queue_retry_minutes"));
+    xs *ntid = tid(retries * 60 * qrt);
     xs *fn   = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
     xs *tfn  = xs_fmt("%s.tmp", fn);
     FILE *f;
 
     if ((f = fopen(tfn, "w")) != NULL) {
         xs *qmsg = xs_dict_new();
+        xs *rn   = xs_number_new(retries);
         xs *j;
 
-        qmsg = xs_dict_append(qmsg, "type",   "input");
-        qmsg = xs_dict_append(qmsg, "object", msg);
-        qmsg = xs_dict_append(qmsg, "req",    req);
+        qmsg = xs_dict_append(qmsg, "type",    "input");
+        qmsg = xs_dict_append(qmsg, "object",  msg);
+        qmsg = xs_dict_append(qmsg, "req",     req);
+        qmsg = xs_dict_append(qmsg, "retries", rn);
 
         j = xs_json_dumps_pp(qmsg, 4);
 
diff --git a/snac.h b/snac.h
index 60322d4..c65b222 100644
--- a/snac.h
+++ b/snac.h
@@ -83,7 +83,7 @@ int actor_get(snac *snac, char *actor, d_char **data);
 
 int static_get(snac *snac, char *id, d_char **data, int *size);
 
-void enqueue_input(snac *snac, char *msg, char *req);
+void enqueue_input(snac *snac, char *msg, char *req, int retries);
 void enqueue_output(snac *snac, char *msg, char *actor, int retries);
 
 d_char *queue(snac *snac);