3 * tailing reliable realtime streaming feeder for inn
4 * recv.c - receiving peer responses and disposing of articles
6 * Copyright Ian Jackson <ijackson@chiark.greenend.org.uk>
7 * and contributors; see LICENCE.txt.
8 * SPDX-License-Identifier: GPL-3.0-or-later
13 /*========== handling responses from peer ==========*/
15 const oop_rd_style peer_rd_style= {
16 OOP_RD_DELIM_STRIP, '\n',
18 OOP_RD_SHORTREC_FORBID
21 void *peer_rd_err(oop_source *lp, oop_read *oread, oop_rd_event ev,
22 const char *errmsg, int errnoval,
23 const char *data, size_t recsz, void *conn_v) {
25 connfail(conn, "error receiving from peer: %s", errmsg);
29 static Article *article_reply_check(Conn *conn, const char *response,
30 int code_indicates_streaming,
32 /* 1:yes, -1:no, 0:dontcare */,
33 const char *sanitised_response) {
34 Article *art= LIST_HEAD(conn->sent);
38 "peer gave unexpected response when no commands outstanding: %s",
43 if (code_indicates_streaming) {
44 assert(!memchr(response, 0, 4)); /* ensured by peer_rd_ok */
46 connfail(conn, "peer gave streaming response code "
47 " to IHAVE or subsequent body: %s", sanitised_response);
50 const char *got_mid= response+4;
51 int got_midlen= strcspn(got_mid, " \n\r");
52 if (got_midlen<3 || got_mid[0]!='<' || got_mid[got_midlen-1]!='>') {
53 connfail(conn, "peer gave streaming response with syntactically invalid"
54 " messageid: %s", sanitised_response);
57 if (got_midlen != art->midlen ||
58 memcmp(got_mid, art->messageid, got_midlen)) {
59 connfail(conn, "peer gave streaming response code to wrong article -"
60 " probable synchronisation problem; we offered: %s;"
62 art->messageid, sanitised_response);
67 connfail(conn, "peer gave non-streaming response code to"
68 " CHECK/TAKETHIS: %s", sanitised_response);
73 if (must_have_sent>0 && art->state < art_Wanted) {
74 connfail(conn, "peer says article accepted but"
75 " we had not sent the body: %s", sanitised_response);
78 if (must_have_sent<0 && art->state >= art_Wanted) {
79 connfail(conn, "peer says please sent the article but we just did: %s",
84 Article *art_again= LIST_REMHEAD(conn->sent);
85 assert(art_again == art);
89 static void update_nocheck(int accepted) {
90 accept_proportion *= nocheck_decay;
91 accept_proportion += accepted * (1.0 - nocheck_decay);
92 int new_nocheck= accept_proportion >= nocheck_thresh;
93 if (new_nocheck && !nocheck_reported) {
94 notice("entering nocheck mode for the first time");
96 } else if (new_nocheck != nocheck) {
97 dbg("nocheck mode %s", new_nocheck ? "start" : "stop");
102 void article_done(Article *art, int whichcount) {
103 if (whichcount>=0 && !art->missing)
104 art->ipf->counts.results[art->state][whichcount]++;
106 if (whichcount == RC_accepted)
108 else if (whichcount == RC_unwanted ||
109 (whichcount == RC_rejected && art->state == art_Unsolicited))
112 InputFile *ipf= art->ipf;
114 while (art->blanklen) {
115 static const char spaces[]=
125 int nspaces= sizeof(spaces)-1;
126 int w= art->blanklen; if (w > nspaces) w= nspaces;
127 int r= pwrite(ipf->fd, spaces, w, art->offset);
129 if (errno==EINTR) continue;
130 syscrash("failed to blank entry for %s (length %d at offset %lu) in %s",
131 art->messageid, art->blanklen,
132 (unsigned long)art->offset, ipf->path);
134 assert(r>=0 && r<=w);
140 assert(ipf->inprogress >= 0);
143 if (!ipf->inprogress && ipf != main_input_file)
144 queue_check_input_done();
147 void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev,
148 const char *errmsg, int errnoval,
149 const char *data, size_t recsz, void *conn_v) {
152 if (ev == OOP_RD_EOF) {
153 connfail(conn, "unexpected EOF from peer");
156 assert(ev == OOP_RD_OK);
158 char *sani= sanitise(data,-1);
161 unsigned long code= strtoul(data, &ep, 10);
162 if (ep != data+3 || *ep != ' ' || data[0]=='0') {
163 connfail(conn, "badly formatted response from peer: %s", sani);
167 int busy= conn_busy(conn);
169 if (conn->quitting) {
170 if (code!=205 && code!=400) {
171 connfail(conn, "peer gave unexpected response to QUIT (%s): %s",
172 conn->quitting, sani);
174 LIST_REMOVE(conns,conn);
175 info("C%d (now %d) idle connection closed (%s)",
176 conn->fd, conns.count, conn->quitting);
177 notice_conns_fewer();
184 conn->since_activity= 0;
187 #define GET_ARTICLE(musthavesent) do{ \
188 art= article_reply_check(conn, data, code_streaming, musthavesent, sani); \
189 if (!art) return OOP_CONTINUE; /* reply_check has failed the conn */ \
192 #define ARTICLE_DEALTWITH(streaming,musthavesent,how) do{ \
193 code_streaming= (streaming); \
194 GET_ARTICLE(musthavesent); \
195 article_done(art, RC_##how); \
199 #define PEERBADMSG(m) do { \
200 connfail(conn, m ": %s", sani); return OOP_CONTINUE; \
203 int code_streaming= 0;
207 default: PEERBADMSG("peer sent unexpected message");
211 PEERBADMSG("peer timed us out or stopped accepting articles");
213 LIST_REMOVE(conns,conn);
214 info("C%d (now %d) idle connection closed by peer",
215 conns.count, conn->fd);
216 notice_conns_fewer();
220 case 435: ARTICLE_DEALTWITH(0,0,unwanted); /* IHAVE says they have it */
221 case 438: ARTICLE_DEALTWITH(1,0,unwanted); /* CHECK/TAKETHIS: they have it */
223 case 235: ARTICLE_DEALTWITH(0,1,accepted); /* IHAVE says thanks */
224 case 239: ARTICLE_DEALTWITH(1,1,accepted); /* TAKETHIS says thanks */
226 case 437: ARTICLE_DEALTWITH(0,0,rejected); /* IHAVE says rejected */
227 case 439: ARTICLE_DEALTWITH(1,0,rejected); /* TAKETHIS says rejected */
229 case 238: /* CHECK says send it */
231 case 335: /* IHAVE says send it */
233 assert(art->state == art_Unchecked);
234 art->ipf->counts.results[art->state][RC_accepted]++;
235 art->state= art_Wanted;
236 LIST_ADDTAIL(conn->priority, art);
239 case 431: /* CHECK or TAKETHIS says try later */
241 case 436: /* IHAVE says try later */
243 article_defer(art, RC_deferred);
249 conn_maybe_write(conn);
250 check_assign_articles();