monkee.ch/snac2
1
0
Fork 0
mirror of https://codeberg.org/grunfink/snac2.git synced 2024-11-14 17:45:04 +00:00
Code Issues Projects Releases Packages Wiki Activity

The input queue also has retries.

Browse source
This commit is contained in:
default 2022-09-28 20:41:07 +02:00
parent 51ed24be47
commit 2be2c07e9c
3 changed files with 24 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
activitypub.c
View file

@ -557,7 +557,7 @@ d_char *msg_note(snac *snac, char *content, char *rcpts, char *in_reply_to)
/** queues **/ /** 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 */ /* processes an ActivityPub message from the input queue */
{ {
/* actor and type exist, were checked previously */ /* actor and type exist, were checked previously */
@ -575,10 +575,8 @@ void process_message(snac *snac, char *msg, char *req)
/* bring the actor */ /* bring the actor */
if (!valid_status(actor_request(snac, actor, &actor_o))) { 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)); snac_log(snac, xs_fmt("error requesting actor %s -- retry later", actor));
return; return 0;
} }
/* check the signature */ /* check the signature */
@ -690,6 +688,8 @@ void process_message(snac *snac, char *msg, char *req)
*/ */
else else
snac_debug(snac, 1, xs_fmt("process_message type '%s' ignored", type)); 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 */ /* process the message */
char *msg = xs_dict_get(q_item, "object"); char *msg = xs_dict_get(q_item, "object");
char *req = xs_dict_get(q_item, "req"); 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)) { if (valid_status(status)) {
enqueue_input(&snac, msg, req); enqueue_input(&snac, msg, req, 0);
*ctype = "application/activity+json"; *ctype = "application/activity+json";
} }

13
data.c
View file

@ -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 */ /* 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 *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
xs *tfn = xs_fmt("%s.tmp", fn); xs *tfn = xs_fmt("%s.tmp", fn);
FILE *f; FILE *f;
if ((f = fopen(tfn, "w")) != NULL) { if ((f = fopen(tfn, "w")) != NULL) {
xs *qmsg = xs_dict_new(); xs *qmsg = xs_dict_new();
xs *rn = xs_number_new(retries);
xs *j; xs *j;
qmsg = xs_dict_append(qmsg, "type", "input"); qmsg = xs_dict_append(qmsg, "type", "input");
qmsg = xs_dict_append(qmsg, "object", msg); qmsg = xs_dict_append(qmsg, "object", msg);
qmsg = xs_dict_append(qmsg, "req", req); qmsg = xs_dict_append(qmsg, "req", req);
qmsg = xs_dict_append(qmsg, "retries", rn);
j = xs_json_dumps_pp(qmsg, 4); j = xs_json_dumps_pp(qmsg, 4);

2
snac.h
View file

@ -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); 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); void enqueue_output(snac *snac, char *msg, char *actor, int retries);
d_char *queue(snac *snac); d_char *queue(snac *snac);