diff options
| -rw-r--r-- | activitypub.c | 21 | ||||
| -rw-r--r-- | data.c | 13 | ||||
| -rw-r--r-- | snac.h | 2 |
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"; } @@ -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); @@ -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); |