/* Distributed Checksum Clearinghouse
 *
 * work on a job in the server
 *
 * Copyright (c) 2008 by Rhyolite Software, LLC
 *
 * This agreement is not applicable to any entity which sells anti-spam
 * solutions to others or provides an anti-spam solution as part of a
 * security solution sold to other entities, or to a private network
 * which employs the DCC or uses data provided by operation of the DCC
 * but does not provide corresponding data to other users.
 *
 * Permission to use, copy, modify, and distribute this software without
 * changes for any purpose with or without fee is hereby granted, provided
 * that the above copyright notice and this permission notice appear in all
 * copies and any distributed versions or copies are either unchanged
 * or not called anything similar to "DCC" or "Distributed Checksum
 * Clearinghouse".
 *
 * Parties not eligible to receive a license under this agreement can
 * obtain a commercial license to use DCC by contacting Rhyolite Software
 * at sales@rhyolite.com.
 *
 * A commercial license would be for Distributed Checksum and Reputation
 * Clearinghouse software.  That software includes additional features.  This
 * free license for Distributed ChecksumClearinghouse Software does not in any
 * way grant permision to use Distributed Checksum and Reputation Clearinghouse
 * software
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND RHYOLITE SOFTWARE, LLC DISCLAIMS ALL
 * WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL RHYOLITE SOFTWARE, LLC
 * BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES
 * OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
 * WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION,
 * ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
 * SOFTWARE.
 *
 * Rhyolite Software DCC 1.3.103-1.287 $Revision$
 */

#include "dccd_defs.h"

typedef struct {
    time_t us;
    u_int   ops;
} Q_DELAY_SEC;
static Q_DELAY_SEC q_delays[9];
static Q_DELAY_SEC q_delays_sum;	/* sum of all but q_delays[0] */
static time_t q_delays_start;		/* second for q_delayw[0] */

DCCD_STATS dccd_stats;

u_char query_only;			/* 1=treat reports as queries */

u_char grey_weak_body;			/* 1=ignore bodies for greylisting */
u_char grey_weak_ip;			/* 1=a good triple whitelists addr */

static u_char ridc_get(QUEUE *);


/* report cache used to detect duplicate or retransmitted reports */
static RIDC *ridc_newest, *ridc_oldest;
static RIDC **ridc_hash;
static int ridc_hash_len;


static inline RIDC **
ridc_hash_fnc(DCC_HDR *hdr)
{
	u_int32_t sum;

	/* The client's (ID,RID,HID,PID) should be unique and constant for
	 * retransmissions of a single request.  It should make a reasonable
	 * hash value.  We cannot trust it entirely, if only because of
	 * anonymous clients */
	sum = hdr->sender;
	sum += hdr->op_nums.h;
	sum += hdr->op_nums.p;
	sum += hdr->op_nums.r;

	return &ridc_hash[mhash(sum, ridc_hash_len)];
}



static void
ridc_ref(RIDC *ridc)
{
	ridc->last_used = db_time.tv_sec;
	if (!ridc->newer)
		return;			/* it's already newest */

	ridc->newer->older = ridc->older;
	if (ridc->older)
		ridc->older->newer = ridc->newer;
	else
		ridc_oldest = ridc->newer;
	ridc->older = ridc_newest;
	ridc->newer = 0;
	ridc_newest->newer = ridc;
	ridc_newest = ridc;
}



/* get a free report cache block */
static RIDC *
ridc_get_free(void)
{
	RIDC *ridc;
	time_t stale = db_time.tv_sec - DCC_MAX_RETRANS_DELAY_SECS;

	for (ridc = ridc_oldest; ridc != 0; ridc = ridc->newer) {
		if (ridc->last_used < stale) {
			/* found one, so recycle it */
			if (ridc->fwd)
				ridc->fwd->bak = ridc->bak;
			if (ridc->bak)
				ridc->bak->fwd = ridc->fwd;
			else if (ridc->hash)
				*ridc->hash = ridc->fwd;
			ridc->bak = 0;
			ridc_ref(ridc);
			return ridc;
		}
	}

	/* there are no free blocks that are old enough to recycle */
	return 0;
}



/* make some (more) RID blocks and (re)build the hash table */
static void
ridc_make(void)
{
	int new_len, old_len, j;
	RIDC *ridc, *ridc2, **ridch, **old_ridc_hash;

	new_len = queue_max;
	ridc = dcc_malloc(new_len*sizeof(*ridc));
	if (!ridc)
		dcc_logbad(EX_OSERR, "malloc(%d RIDC blocks) failed",
			   new_len);
	memset(ridc, 0, new_len*sizeof(*ridc));
	for (j = 0; j < new_len; ++j, ++ridc) {	/* make the new blocks oldest */
		if (!ridc_oldest) {
			ridc_oldest = ridc_newest = ridc;
		} else {
			ridc_oldest->older = ridc;
			ridc->newer = ridc_oldest;
			ridc_oldest = ridc;
		}
	}

	/* rebuild and expand the hash table */
	old_len = ridc_hash_len;
	ridc_hash_len += new_len;
	old_ridc_hash = ridc_hash;
	ridc_hash = dcc_malloc(ridc_hash_len*sizeof(*ridch));
	if (!ridc_hash)
		dcc_logbad(EX_OSERR, "malloc(%d RIDC hash table) failed",
			   ridc_hash_len);
	memset(ridc_hash, 0, ridc_hash_len*sizeof(*ridch));
	if (old_len != 0) {
		do {
			for (ridc = old_ridc_hash[--old_len];
			     ridc != 0;
			     ridc = ridc2) {
				ridch = ridc_hash_fnc(&ridc->hdr);
				ridc2 = ridc->fwd;
				ridc->bak = 0;
				ridc->hash = ridch;
				if ((ridc->fwd = *ridch) != 0)
					ridc->fwd->bak = ridc;
				*ridch = ridc;
			}
		} while (old_len != 0);
		dcc_free(old_ridc_hash);
	}
}



/* get the report cache block for an operation */
static u_char				/* 0=new operation, 1=retransmission */
ridc_get(QUEUE *q)
{
	RIDC *ridc, **ridch;

	for (;;) {
		if (ridc_hash) {
			/* look for the existing report cache block */
			ridch = ridc_hash_fnc(&q->pkt.hdr);
			for (ridc = *ridch; ridc; ridc = ridc->fwd) {
				/* Reports are relatively small, so we
				 * can afford to not trust the client's
				 * RID to be unique.  Compare all but the
				 * client's transmission #. 
				 * Also check client's UDP port # because
				 * it should be unchanged regardless of
				 * multi-homing. */
				if (ridc->clnt_port == q->clnt_su.ipv4.sin_port
				    && !memcmp(&ridc->hdr, &q->pkt.hdr,
					       sizeof(ridc->hdr)
					       - sizeof(ridc->hdr.op_nums.t))) {
					/* found it, so make it newest */
					ridc_ref(ridc);
					q->ridc = ridc;
					return 1;
				}
			}

			/* the block does not already exist, so create it */
			ridc = ridc_get_free();
			if (ridc)
				break;
		}
		/* we are out of report cache blocks, so make more */
		ridc_make();

		/* re-hash because our previous pointer is invalid */
	}

	memcpy(&ridc->hdr, &q->pkt.hdr, sizeof(ridc->hdr));
	ridc->clnt_port = q->clnt_su.ipv4.sin_port;
	ridc->op = DCC_OP_INVALID;
	ridc->bad = 1;
	ridc->len = 0;
	ridc->hash = ridch;
	ridc->fwd = *ridch;
	if (ridc->fwd)
		ridc->fwd->bak = ridc;
	*ridch = ridc;

	q->ridc = ridc;
	return 0;
}



#define RIDC_BAD(q) {if ((q)->ridc) (q)->ridc->bad = 1;}


/* update the average queue delay at the start of a new second */
static void
update_q_delay(void)
{
	time_t secs;
	Q_DELAY_SEC *src, *tgt;

	secs = db_time.tv_sec - q_delays_start;
	if (secs == 0)
		return;

	/* At the start of a new second,
	 * forget the delays for old seconds we no longer care about
	 * and start accumulating delays for the new second
	 * Slide accumulated delays and total operations previous seconds. */
	q_delays_start = db_time.tv_sec;
	q_delays_sum.us = 0;
	q_delays_sum.ops = 0;
	tgt = LAST(q_delays);
	if (secs > 0 && secs < DIM(q_delays)) {
		src = tgt - secs;
		do {
			q_delays_sum.us += (tgt->us = src->us);
			q_delays_sum.ops += (tgt->ops = src->ops);
			--tgt;
		} while (src-- != &q_delays[0]);
	}
	memset(q_delays, 0, sizeof(q_delays[0]) * (tgt+1 - q_delays));
}



/* compute the average queue delay this client should see */
static u_int
avg_q_delay_ms(const QUEUE *q)
{
	u_int ops;
	time_t us;

	/* get the average service delay excluding per-client-ID delays */
	update_q_delay();
	ops = q_delays[0].ops + q_delays_sum.ops;
	if (ops == 0)
		us = 0;
	else
		us = (q_delays[0].us + q_delays_sum.us + ops/2) / ops;

	/* add the per-client-ID penalty */
	us += q->delay_us;
	return (us + 500) / 1000;
}



/* get a unique timestamp */
static void
get_ts(DCC_TS *ts)			/* put it here */
{
	static struct timeval prev_time;
	static int faked;

	/* if we have generated a lot of fake timestamps
	 * and our snapshot of the clock is old,
	 * then check the clock in the hope it has ticked */
	if (db_time.tv_usec <= prev_time.tv_usec
	    && db_time.tv_sec == prev_time.tv_sec
	    && faked > 100) {
		faked = 0;
		gettimeofday(&db_time, 0);
	}

	/* Try to make the next timestamp unique, but only as long
	 * as time itself marches forward.  This must work many times
	 * a second, or the resoltion of DCC timestaps.
	 * Worse, the increment can exhaust values from future seconds.
	 * Forget about it if the problem lasts for more than 5 minutes. */
	if (db_time.tv_sec > prev_time.tv_sec
	    || (db_time.tv_sec == prev_time.tv_sec
		&& db_time.tv_usec > prev_time.tv_usec)
	    || db_time.tv_sec < prev_time.tv_sec-5*60) {
		/* either the current time is good enough or we must
		 * give up and use it to make the timestamp */
		prev_time = db_time;
		faked = 0;

	} else {
		/* fudge the previous timestamp to make it good enough */
		prev_time.tv_usec += DCC_TS_US_MULT;
		if (prev_time.tv_usec >= DCC_US) {
			prev_time.tv_usec -= DCC_US;
			++prev_time.tv_sec;
		}
		++faked;
	}

	dcc_timeval2ts(ts, &prev_time, 0);
}



/* find database record for a server-ID
 *	use only db_sts.hash and db_sts.rcd2
 *	put the result in db_sts.rcd2 */
int					/* -1=broken database 0=no record */
find_srvr_rcd(const DCC_SUM sum, const char *str)
{
	DB_RCD_CK *found_ck;
	DB_PTR prev;
	int failsafe;

	switch (db_lookup(dcc_emsg, DCC_CK_SRVR_ID, sum,
			  0, MAX_HASH_ENTRIES,
			  &db_sts.hash, &db_sts.rcd2, &found_ck)) {
	case DB_FOUND_LATER:
	case DB_FOUND_SYSERR:
		DB_ERROR_MSG2(str, dcc_emsg);
		return -1;
	case DB_FOUND_IT:
		/* look for a record that is neither obsolete nor deleted */
		for (failsafe = 0; failsafe < 20; ++failsafe) {
			if (!DB_CK_OBS(found_ck)
			    && DB_TGTS_RCD(db_sts.rcd2.d.r) != 0)
				return 1;
			prev = DB_PTR_EX(found_ck->prev);
			if (prev == DB_PTR_NULL)
				return 0;
			found_ck = db_map_rcd_ck(dcc_emsg, &db_sts.rcd2, prev,
						 DCC_CK_SRVR_ID);
			if (!found_ck) {
				DB_ERROR_MSG2(str, dcc_emsg);
				return -1;
			}
		}
		break;
	case DB_FOUND_EMPTY:
	case DB_FOUND_CHAIN:
	case DB_FOUND_INTRUDER:
		break;
	}
	return 0;
}



/* find the database record of the type of a server
 *	use only db_sts.hash and db_sts.rcd2
 *	put the result in db_sts.rcd2 */
int					/* -1=broken database 0=no record */
find_srvr_rcd_type(DCC_SRVR_ID tgt_id)
{
	DCC_SUM srvr_id_sum;

	if (db_failed_line)
		return -1;
	memset(srvr_id_sum, 0, sizeof(srvr_id_sum));
	srvr_id_sum[0] = DCC_CK_SRVR_ID;
	srvr_id_sum[1] = tgt_id >> 8;
	srvr_id_sum[2] = tgt_id;
	return find_srvr_rcd(srvr_id_sum, "checking server-ID state");
}



/* find the server type in the table of IDs */
ID_TBL *
find_srvr_type(DCC_SRVR_ID tgt_id)
{
	ID_TBL *tp;
	DCC_SRVR_ID srvr_type;

	tp = find_id_tbl(tgt_id);
	if (!tp) {
		/* check the database if it is not in the table */
		if (0 >= find_srvr_rcd_type(tgt_id)) {
			/* assume it is a simple server if there is
			 * no declaration in the database */
			srvr_type = DCC_ID_SRVR_SIMPLE;
		} else {
			srvr_type = DB_RCD_ID(db_sts.rcd2.d.r);
			if (!DCC_ID_SRVR_TYPE(srvr_type))
				srvr_type = DCC_ID_SRVR_SIMPLE;
			/* the free code knows nothing about reputations */
			if (srvr_type == DCC_ID_SRVR_REP_OK)
				srvr_type = DCC_ID_SRVR_SIMPLE;
		}
		/* cache it in the table */
		tp = add_id_tbl(tgt_id);
		tp->srvr_type = srvr_type;
	}
	return tp;
}



/* refresh our claim to our server-ID or similar
 *	use only db_sts.hash and db_sts.rcd2 */
void
refresh_srvr_rcd(const DCC_SUM sum, DCC_SRVR_ID val, const char *str)
{
	DCC_TS old;
	DB_RCD rcd;
	int i;

	/* add a new record
	 * only if no recent record of the right value exists */
	i = find_srvr_rcd(sum, str);
	if (i < 0)
		return;			/* broken database */
	if (i > 0
	    && DB_RCD_ID(db_sts.rcd2.d.r) == val) {
		dcc_timeval2ts(&old, &db_time, -DCC_SRVR_ID_SECS/2);
		if (!dcc_ts_older_ts(&db_sts.rcd2.d.r->ts, &old))
			return;
	}

	memset(&rcd, 0, sizeof(rcd));
	get_ts(&rcd.ts);
	rcd.srvr_id_auth = val;
	DB_TGTS_RCD_SET(&rcd, 1);
	rcd.fgs_num_cks = 1;
	rcd.cks[0].type_fgs = DCC_CK_SRVR_ID;
	memcpy(rcd.cks[0].sum, sum, sizeof(DCC_SUM));
	if (!db_add_rcd(dcc_emsg, &rcd))
		DB_ERROR_MSG2(str, dcc_emsg);
}



static void
send_resp(const QUEUE *q,
	  DCC_HDR *hdr,			/* length in host byte order */
	  u_char no_msg)
{
	u_int save_len;
	char ob[DCC_OPBUF];
	int len, i;

	len = hdr->len;
	hdr->len = htons(len);
	/* callers must have dealt with the variations due to versions */
	if (q->pkt.hdr.pkt_vers < DCC_PKT_VERSION_MIN)
		hdr->pkt_vers = DCC_PKT_VERSION_MIN;
	else if (q->pkt.hdr.pkt_vers > DCC_PKT_VERSION_MAX)
		hdr->pkt_vers = DCC_PKT_VERSION_MAX;
	else
		hdr->pkt_vers = q->pkt.hdr.pkt_vers;
	hdr->sender = htonl(my_srvr_id);
	hdr->op_nums = q->pkt.hdr.op_nums;
	if (q->passwd[0] != '\0') {
		/* sign with the password that authenticated the client */
		dcc_sign(q->passwd, sizeof(q->passwd), hdr, len);
#ifdef DCC_PKT_VERSION8
	} else if (q->pkt.hdr.pkt_vers <= DCC_PKT_VERSION8) {
		/* Sign old protocol responses with the client's transaction
		 * numbers if we do not have a good password.
		 * This happens with anonymous clients */
		dcc_sign((char *)&q->pkt.hdr.op_nums,
			 sizeof(q->pkt.hdr.op_nums),
			 hdr, len);
#endif
	} else {
		memset((char *)hdr + (len-sizeof(DCC_SIGNATURE)), 0,
		       sizeof(DCC_SIGNATURE));
	}

	if (q->ridc) {
		save_len = len-sizeof(*hdr)-sizeof(DCC_SIGNATURE);
		if (save_len > ISZ(q->ridc->result)) {
			if (hdr->op == DCC_OP_ERROR)
				save_len = sizeof(q->ridc->result);
			else
				dcc_logbad(EX_SOFTWARE, "RIDC buffer overflow");
		}
		q->ridc->len = save_len;
		memcpy(&q->ridc->result, hdr+1, save_len);
		q->ridc->op = hdr->op;
		q->ridc->bad = 0;
	}

	i = sendto(q->sp->udp, hdr, len, 0,
		   &q->clnt_su.sa, DCC_SU_LEN(&q->clnt_su));
	if (i < 0) {
		clnt_msg(q, "sendto(%s, %s): %s",
			 dcc_hdr_op2str(ob, sizeof(ob), hdr), Q_CIP(q),
			 ERROR_STR());
	} else if (len != i) {
		clnt_msg(q, "sendto(%s, %s)=%d instead of %d",
			 dcc_hdr_op2str(ob, sizeof(ob), hdr), Q_CIP(q),
			 i, len);
	} else if (!no_msg
		   && (dccd_tracemask & ((hdr->op == DCC_OP_ANSWER
					  || hdr->op == DCC_OP_NOP)
					 ? DCC_TRACE_QUERY_BIT
					 : DCC_TRACE_ADMN_BIT))) {
		dcc_trace_msg("sent %s to %s for %s",
			      dcc_hdr_op2str(ob, sizeof(ob), hdr),
			      Q_CIP(q), qop2str(q));
	}

}



/* do not send an error response to a client */
static void PATTRIB(2,3)
forget_error(const QUEUE *q, const char *p, ...)
{
	va_list args;

	RIDC_BAD(q);

	if ((!q->flags & Q_FG_BAD_PASSWD)
	    && !(q->rl->d.flags & RL_FG_BLS)) {
		q->rl->d.flags |= RL_FG_BL_BAD;
		++dccd_stats.bad_op;
	}

	va_start(args, p);
	vclnt_msg(q, p, args);
	va_end(args);
}



/* send an error response to a client */
static void
send_error(const QUEUE *q, const char *p, ...)
{
	DCC_ERROR buf;
	int slen;
	va_list args;


	/* build and log the message */
	va_start(args, p);
	slen = vsnprintf(buf.msg, sizeof(buf.msg), p, args);
	if (slen > ISZ(buf.msg)-1)
		slen = ISZ(buf.msg)-1;
	va_end(args);
	clnt_msg(q, "\"%s\" sent to %s", buf.msg, Q_CIP(q));

	/* send it */
	buf.hdr.len = sizeof(buf)-sizeof(buf.msg)+slen+1;
	buf.hdr.op = DCC_OP_ERROR;
	send_resp(q, &buf.hdr, 1);

	++dccd_stats.send_error;
}



#define NORESP_EMSG(q) noresp_emsg(q, __LINE__)

static void
noresp_emsg(const QUEUE *q, int linenum)
{
	dcc_error_msg("error near line %d in "DCC_VERSION" "__FILE__, linenum);
	RIDC_BAD(q);
}



/* tell client that a NOP or an administrative request was ok */
static void
send_ok(QUEUE *q)
{
	DCC_OK buf;
	time_t us;

	memset(&buf, 0, sizeof(buf));

	buf.max_pkt_vers = max(min(q->pkt.hdr.pkt_vers, DCC_PKT_VERSION_MAX),
			       DCC_PKT_VERSION_MIN);
	us = (q->delay_us + 500) / 1000;
	buf.qdelay_ms = htons(us);
	strncpy(buf.brand, brand, sizeof(buf.brand));
	buf.hdr.op = DCC_OP_OK;
	buf.hdr.len = sizeof(buf);

	send_resp(q, &buf.hdr, 0);
}



static void
repeat_resp(QUEUE *q)
{
	struct {
	    DCC_HDR hdr;
	    u_char  b[sizeof(q->ridc->result)];
	} buf;
	char ob[DCC_OPBUF];

	++dccd_stats.report_retrans;

	if (q->ridc->bad) {
		TMSG1(RIDC, "repeat drop of %s", from_id_ip(q, 1));
		return;
	}

	memcpy(&buf.hdr+1, &q->ridc->result, q->ridc->len);
	buf.hdr.op = q->ridc->op;
	buf.hdr.len = htons(q->ridc->len
			    + sizeof(buf.hdr) + sizeof(DCC_SIGNATURE));
	TMSG2(RIDC, "repeat previous answer of %s for %s",
	      dcc_hdr_op2str(ob, sizeof(ob), &buf.hdr),
	      from_id_ip(q, 1));
	buf.hdr.len = ntohs(buf.hdr.len);
	send_resp(q, &buf.hdr, 0);
}



/* find a checksum in the database
 *	use only db_sts.hash and db_sts.rcd2
 *	put the result in db_sts.rcd2 */
static u_char				/* 0=broken database */
get_ck_tgts(DCC_TGTS *tgtsp,
	    const DB_RCD_CK **pfound_ck,
	    u_char must_have_it,	/* 1=database broken if cksum absent */
	    DCC_CK_TYPES type,
	    const DCC_SUM sum)
{
	DB_RCD_CK *found_ck;

	switch (db_lookup(dcc_emsg, type, sum, 0, MAX_HASH_ENTRIES,
			  &db_sts.hash, &db_sts.rcd2, &found_ck)) {
	case DB_FOUND_LATER:
	case DB_FOUND_SYSERR:
		DB_ERROR_MSG(dcc_emsg);
		return 0;
	case DB_FOUND_IT:
		if (pfound_ck)
			*pfound_ck = found_ck;
		if (tgtsp)
			*tgtsp = DB_TGTS_CK(found_ck);
		break;
	case DB_FOUND_EMPTY:
	case DB_FOUND_CHAIN:
	case DB_FOUND_INTRUDER:
		if (must_have_it) {
			db_error_msg(__LINE__,__FILE__,
				     "missing hash entry for %s %s ",
				     DB_TYPE2STR(type),
				     dcc_ck2str_err(type, sum, 0));
			return 0;
		}
		if (pfound_ck)
			*pfound_ck = 0;
		if (tgtsp)
			*tgtsp = 0;
		break;
	}
	return 1;
}



/* see if a count just passed a multiple of a threshold and so is
 *	worth flooding or summarizing */
static u_char				/* 1=time to summarize this checksum */
quick_sum_thold(DCC_CK_TYPES type,
		DCC_TGTS rpt_tgts,	/* targets in this report */
		DCC_TGTS ck_tgts)	/* grand total */
{
	static DCC_TGTS thold_mults[] = {
	    1, 2, 3, 5, 10
	};
	DCC_TGTS thold;
	DCC_TGTS mult, new_mult, old_mult;
	int i;

	thold = flod_tholds[type];
	if (ck_tgts < thold
	    || thold >= DCC_TGTS_TOO_MANY)
		return 0;
	if (thold == 0)
		return 1;

	new_mult = ck_tgts / thold;
	old_mult = (ck_tgts - rpt_tgts) / thold;
	for (i = 0; i < DIM(thold_mults); ++i) {
		mult = thold_mults[i];
		if (old_mult < mult)
			return (new_mult >= mult);
	}
	return 0;
}



/* compute summarizable total for one checksum
 *	use  db_sts.hash, db_sts.rcd2, and *rcd_st */
static DCC_TGTS				/* DCC_TGTS_INVALID=broken database */
sum_total(DCC_CK_TYPES type,		/* look for this */
	  const DCC_SUM sum,
	  u_char must_have_it,
	  DB_STATE *rcd_st,		/* starting here */
	  const DB_RCD_CK *found_ck,
	  u_char *undelay_ok,		/* 0=cannot undelay by clearing bit */
	  DB_PTR *sum_oldest)
{
	DB_PTR prev;
	DCC_TGTS rcd_tgts, sub_total;
	int limit;

	if (!rcd_st) {
		if (!get_ck_tgts(0, &found_ck, must_have_it, type, sum))
			return DCC_TGTS_INVALID;
		if (!found_ck)
			return 0;
		rcd_st = &db_sts.rcd2;
	}

	if (sum_oldest)
		*sum_oldest = DB_PTR_MAX;
	sub_total = 0;
	for (limit = 10000; limit >= 0; --limit) {
		/* stop adding reports at the first summary or
		 * compressed record in the hash chain */
		if (DB_RCD_SUMRY(rcd_st->d.r)
		    || DB_RCD_ID(rcd_st->d.r) == DCC_ID_COMP)
			break;

		/* honor deletions */
		rcd_tgts = DB_TGTS_RCD(rcd_st->d.r);
		if (rcd_tgts == 0)
			break;

		/* We can only summarize our own delayed reports
		 * to keep loops in the flooding topology from
		 * inflating totals. */
		if (DB_RCD_DELAY(rcd_st->d.r)
		    && DB_RCD_ID(rcd_st->d.r) == my_srvr_id) {
			if (sum_oldest)
				*sum_oldest = rcd_st->s.rptr;
			sub_total = db_sum_ck(sub_total, rcd_tgts, type);
			/* if we summarize more than one record,
			 * then we cannot simply convert the record */
			if (undelay_ok
			    && db_sts.sumrcd.s.rptr != rcd_st->s.rptr)
				*undelay_ok = 0;
		}
		prev = DB_PTR_EX(found_ck->prev);
		if (prev == DB_PTR_NULL)
			break;
		rcd_st = &db_sts.rcd2;
		found_ck = db_map_rcd_ck(dcc_emsg, rcd_st, prev, type);
		if (!found_ck) {
			DB_ERROR_MSG(dcc_emsg);
			return 0;
		}
	}

	return sub_total;
}



/* generate a summary record of checksum counts
 *	db_sts.sumrcd points to the record being summarize on entry
 *	    On exit db_sts.sumrcd points to the same record or the original
 *	    has been trashed and db_sts.sumrcd points to a moved copy.
 *	Use db_sts.rcd, db_sts.hash, db_sts.rcd2, db_sts.free, db_sts.tmp */
static u_char				/* 0=sick db, 1=ok, 2=moved rcd */
summarize_rcd(u_char dly)		/* 1=working on delayed records */
{
	DB_RCD new;
	DCC_TGTS rcd_tgts, ck_tgts, new_tgts, sub_total;
	DCC_CK_TYPES type;
	DB_RCD_CK *cur_ck, *new_ck;
	int cur_num_cks;
	u_char ck_needed;		/* 0=junk cksum, 2=needed in new rcd */
	u_char rcd_needed;		/* 1=have created rcd to add */
	u_char undelay_ok;		/* 1=ok to remove delay bit */
	u_char move_ok;
	DB_PTR sum_oldest;
	DB_PTR rcd_pos;

	if (db_lock() < 0)
		return 0;

	/* For each checksum whose flooding was delayed but is now needed,
	 * generate a fake record that will be flooded */
	cur_num_cks = DB_NUM_CKS(db_sts.sumrcd.d.r);
	cur_ck = db_sts.sumrcd.d.r->cks;
	new_tgts = 0;
	undelay_ok = (FLODS_OK()
		      && (DB_RCD_ID(db_sts.sumrcd.d.r) == my_srvr_id));
	move_ok = 1;
	rcd_needed = 0;
	new_ck = new.cks;
	do {
		/* Sum counts of all delayed reports for this checksum */
		type = DB_CK_TYPE(cur_ck);
		if (DB_TEST_NOKEEP(db_parms.nokeep_cks, type))
			continue;

		ck_needed = DB_CK_OBS(cur_ck) ? 0 : 1;

		/* skip trudging through the hash table to find the
		 * most recent instance of the checksum if we
		 * are dealing with a new record and so already
		 * have the most recent instance. */
		sub_total = sum_total(type, cur_ck->sum, 1,
				      !dly ? &db_sts.sumrcd : 0,
				      cur_ck, &undelay_ok, &sum_oldest);
		if (sub_total == DCC_TGTS_INVALID)
			return 0;

		/* Deletions and summaries between our record and the start
		 * of the hash chain remove the need to flood this checkusm */
		if (sub_total == 0) {
			/* skipping a checksum in the original
			 * record makes it impossible to move it */
			move_ok = 0;
			continue;
		}

		if (ck_needed == 1) {
			ck_tgts = DB_TGTS_CK(cur_ck);
			if (dly) {
				/* Flood only 1 summary per delay period */
				if ((flod_mmaps == 0
				     || sum_oldest <= flod_mmaps->delay_pos)
				    && ck_tgts >= flod_tholds[type])
					ck_needed = 2;
			} else {
				/* We are considering the need for a summary
				 * based on a report just received from a client
				 * or by flooding */
				if (quick_sum_thold(type, sub_total, ck_tgts))
					ck_needed = 2;
			}
		}

		if (new_ck != new.cks) {
			/* We have already begun a summary record. */

			if (sub_total == new_tgts) {
				/* extend it with this checksum even if we do
				 * not really need to flood this checksum */
				new_ck->type_fgs = type;
				memcpy(new_ck->sum, cur_ck->sum,
				       sizeof(new_ck->sum));
				++new.fgs_num_cks;
				++new_ck;
				if (ck_needed == 2)
					rcd_needed = 1;
				continue;
			}
			/* We cannot extend the current summary record. */

			/* If we don't really need the checksum,
			 * then forget the checksum. */
			if (ck_needed != 2) {
				/* skipping a checksum in the original
				 * record makes it impossible to move */
				move_ok = 0;
				continue;
			}

			/* Add the current summary record to the database if
			 * it is needed. */
			if (rcd_needed) {
				if (!db_add_rcd(dcc_emsg, &new)) {
					DB_ERROR_MSG(dcc_emsg);
					return 0;
				}
			}
			/* start a new summary with this checksum. */
			rcd_needed = 0;
			/* having added one summary record,
			 * we cannot undelay or move the original record */
			undelay_ok = 0;
		}

		/* start a new summary record */
		new.srvr_id_auth = my_srvr_id;
		get_ts(&new.ts);
		new_tgts = sub_total;
		DB_TGTS_RCD_SET(&new, new_tgts);
		new.fgs_num_cks = DB_RCD_FG_SUMRY+1;
		new_ck = new.cks;
		new_ck->type_fgs = type;
		memcpy(new_ck->sum, cur_ck->sum, sizeof(new_ck->sum));
		++new_ck;
		if (ck_needed == 2)
			rcd_needed = 1;
	} while (++cur_ck, --cur_num_cks > 0);

	/* finished if nothing more to summarize */
	if (!rcd_needed) {
		return 1;
	}

	/* Add the last summary record */
	if (undelay_ok) {
		/* If possible, instead of adding a new record,
		 * change the preceding record to not be delayed
		 * That is possible if the preceding record has
		 * not yet been passed by the flooding */
		if (db_sts.sumrcd.s.rptr >= oflods_max_cur_pos
		    && oflods_max_cur_pos != 0) {
			db_sts.sumrcd.d.r->fgs_num_cks &= ~DB_RCD_FG_DELAY;
			SET_FLUSH_RCD_HDR(&db_sts.sumrcd, 1);
			return 1;
		}

		/* failing that, try to move the record by making a new copy
		 * and deleting the original */
		if (move_ok) {
			/* make the new record */
			memcpy(&new, db_sts.sumrcd.d.r, DB_RCD_LEN(&new));
			new.fgs_num_cks &= ~DB_RCD_FG_DELAY;

			/* delete the old record */
			DB_TGTS_RCD_SET(db_sts.sumrcd.d.r, 0);

			/* adjust the totals in the old record so
			 * that the totals in the new record will be right */
			rcd_tgts = DB_TGTS_RCD(&new);
			cur_num_cks = DB_NUM_CKS(db_sts.sumrcd.d.r);
			cur_ck = db_sts.sumrcd.d.r->cks;
			do {
				new_tgts = DB_TGTS_CK(cur_ck);
				if (new_tgts >= DCC_TGTS_TOO_MANY)
					    continue;
				if (new_tgts != 0)
					new_tgts -= rcd_tgts;
				DB_TGTS_CK_SET(cur_ck, new_tgts);
			}  while (++cur_ck, --cur_num_cks > 0);
			SET_FLUSH_RCD_HDR(&db_sts.sumrcd, 1);

			rcd_pos = db_add_rcd(dcc_emsg, &new);
			if (rcd_pos == DB_PTR_NULL) {
				DB_ERROR_MSG(dcc_emsg);
				return 0;
			}
			if (!db_map_rcd(dcc_emsg, &db_sts.sumrcd, rcd_pos, 0)) {
				DB_ERROR_MSG(dcc_emsg);
				return 0;
			}
			return 1;
		}
	}

	if (!db_add_rcd(dcc_emsg, &new)) {
		DB_ERROR_MSG(dcc_emsg);
		return 0;
	}

	return 1;
}



/* generate a delayed summary for checksums in a record if necessary
 *	The target record is specified by db_sts.sumrcd.  It might be changed
 *	Use db_sts.hash and db_sts.rcd2 */
u_char
summarize_dly(void)
{
	DCC_CK_TYPES type;
	const DB_RCD_CK *cur_ck;
	int cur_num_cks;
	DCC_TGTS ck_tgts;

	/* look for a checksum that could be summarized */
	cur_num_cks = DB_NUM_CKS(db_sts.sumrcd.d.r);
	cur_ck = db_sts.sumrcd.d.r->cks;
	do {
		type = DB_CK_TYPE(cur_ck);
		if (DB_TEST_NOKEEP(db_parms.nokeep_cks, type))
			continue;

		if (!get_ck_tgts(&ck_tgts, 0, 1, type, cur_ck->sum))
			return 0;

		/* nothing to do if the checksum has already been summarized */
		if (DB_RCD_SUMRY(db_sts.rcd2.d.r))
			continue;

		/* spam reports are ignored or not delayed */
		if (ck_tgts == DCC_TGTS_TOO_MANY)
			continue;

		/* Generate a summary for a bulk checksum
		 * Records that are marked "delayed" are not flooded.
		 * If a summary record is not synthesized and if the delay
		 * marking not removed (instead of synthesizing a summary),
		 * then the counts for a checksum will not be flooded. */

		if (ck_tgts >= flod_tholds[type])
			return summarize_rcd(1);
	}  while (++cur_ck, --cur_num_cks > 0);

	return 1;
}



/* See if passing on a flooded report would be worthwhile.  It is worthwhile
 *	to pass on reports of spam that have not been flooded recently
 *	and of checksums that not yet or just barely reached spam.
 *
 *	db_sts.sumrcd points to the new record */
static u_char				/* 0=sick database */
flod_worth(u_char *pflod,		/* set =1 if report should be flooded */
	   const DB_RCD_CK *ck,
	   DCC_CK_TYPES type)
{
	DCC_TS past;
	DCC_TGTS total;
	int limit;
	DB_PTR prev;

	/* if the total with the new report is small,
	 * then we should flood it */
	total = DB_TGTS_CK(ck);
	if (total < REFLOOD_THRESHOLD) {
		/* but only if it is not trivial.
		 * our neighbors should not send trivial reports,
		 * but bugs happen */
		if (total >= BULK_THRESHOLD/2)
			*pflod = 1;
		return 1;
	}

	/* Look for a recent report for this checksum that has been
	 * or will be flooded.  If we find one, and if the total
	 * including it is large enough, we may not need to flood
	 * the incoming report.  If the total is too small, we
	 * must flood the report. */
	dcc_timeval2ts(&past, &db_time, -summarize_delay_secs);
	for (limit = 20; limit >= 0; --limit) {
		prev = DB_PTR_EX(ck->prev);
		if (prev == DB_PTR_NULL)
			break;
		ck = db_map_rcd_ck(dcc_emsg, &db_sts.rcd2, prev, type);
		if (!ck) {
			DB_ERROR_MSG(dcc_emsg);
			return 0;
		}

		/* if the previous total was small,
		 * then we must flood the new report */
		total = DB_TGTS_CK(ck);
		if (total < REFLOOD_THRESHOLD*4) {
			*pflod = 1;
			return 1;
		}

		/* The old total is large.
		 * If this found old report is not very old and good,
		 * we will flood it and so the newest needed not be flooded
		 * and can be marked obsolete. */
		if (!DB_CK_OBS(ck)
		    && dcc_ts_newer_ts(&db_sts.rcd2.d.r->ts, &past))
			return 1;
	}

	/* flood this one if we can't find a recent preceding report */
	*pflod = 1;
	return 1;
}



/* Add a record and deal with delaying its flooding.
 *	We will delay flooding it if its totals are not interesting.
 *	db_sts.sumrcd points to the new record on exit
 *	Use db_sts.rcd, db_sts.hash, db_sts.rcd2, db_sts.free, db_sts.tmp
 *	the database must be locked */
u_char					/* 1=ok, delayed or not, 0=failure */
add_dly_rcd(DB_RCD *new_rcd, u_char flod_in)
{
	DB_PTR rcd_pos;
	int num_cks;
	DB_RCD_CK *new_ck;
	DCC_CK_TYPES type;
	DCC_TGTS rpt_tgts, ck_tgts;
	u_char flod_out;	/* 0=flooded in but not worth flooding out */
	u_char useful = 0;		/* 1=worth delaying */
	u_char summarize = 0;

	/* put the record in the database */
	rcd_pos = db_add_rcd(dcc_emsg, new_rcd);
	if (rcd_pos == DB_PTR_NULL) {
		DB_ERROR_MSG(dcc_emsg);
		return 0;
	}
	if (!db_map_rcd(dcc_emsg, &db_sts.sumrcd, rcd_pos, 0)) {
		DB_ERROR_MSG(dcc_emsg);
		return 0;
	}

	/* delete requests should not be delayed */
	rpt_tgts = DB_TGTS_RCD_RAW(db_sts.sumrcd.d.r);
	if (rpt_tgts == DCC_TGTS_DEL)
		return 1;

	/* we always consider flooding our own reports
	 * and the greylist thresholds are zilch */
	flod_out = !flod_in || grey_on;

	for (num_cks = DB_NUM_CKS(db_sts.sumrcd.d.r),
	     new_ck = db_sts.sumrcd.d.r->cks;
	     num_cks > 0;
	     ++new_ck, --num_cks) {
		/* ingore already obsolete reports of spam */
		if (DB_CK_OBS(new_ck))
			continue;
		/* ignore checksums we won't keep and so won't be flooded */
		type = DB_CK_TYPE(new_ck);
		if (DB_TEST_NOKEEP(db_parms.nokeep_cks, type))
			continue;

		/* Server-ID declarations cannot be summarized and should
		 * not be delayed. */
		if (type == DCC_CK_SRVR_ID) {
			flod_out = 1;
			break;
		}

		ck_tgts = DB_TGTS_CK(new_ck);
		if (ck_tgts == DCC_TGTS_TOO_MANY) {
			/* This checksum has a total of TOO_MANY and so
			 * either the report has a target count of TOO_MANY
			 * or is a report of a checksum already known to
			 * be spam.  Since this report of this checksum
			 * was not marked obsolete as it was linked into the
			 * database, it should not be delayed.  */
			if (rpt_tgts == DCC_TGTS_TOO_MANY) {
				/* if the report is of spam, then all of its
				 * individual checksum totals will be
				 * DCC_TGTS_TOO_MANY.  The checksums will be
				 * obsolete, not kept, or the same as this.
				 * There will be no reputation checksums. */
				return 1;
			}
			/* it is worth sending on even if was not ours */
			flod_out = 1;
			continue;
		}

		/* This report has some potential value and should be delayed
		 * instead of forgotten */
		useful = 1;

		/* Summarize our records for the checksums in this record
		 * if we just passed the threshold for one checksum. */
		if (!summarize
		    && quick_sum_thold(type, rpt_tgts, ck_tgts))
			summarize = 1;

		/* If this is an incoming flooded checksum,
		 * then pass it on if it is novel (has a low total)
		 * or if we have not passed it on recently. */
		if (!flod_out
		    && !flod_worth(&flod_out, new_ck, type))
			return 0;	/* broken database */
	}

	/* Reports that are reports of spam or "trimmed" or "obsolete"
	 * noise should not be summarized or marked to be delayed.
	 * They will be flooded or skipped by the flooder */
	if (!useful)
		return 1;

	if (!flod_in) {
		/* Delay and sooner or later summarize our own
		 * reports of non-spam */
		db_sts.sumrcd.d.r->fgs_num_cks |= DB_RCD_FG_DELAY;

	} else if (!flod_out) {
		/* We are dealing with a report flooded in from another
		 * server that is not (yet?) worth flooding out.
		 * We can't delay it, because we can't delay reports from
		 * other servers, because we cannot summarize them.
		 * Summarizing other servers' reports would allow
		 * loops in the flooding topology to inflate the totals.
		 * So mark it to be expired but not delayed. */
		for (num_cks = DB_NUM_CKS(db_sts.sumrcd.d.r),
		     new_ck = db_sts.sumrcd.d.r->cks;
		     num_cks > 0;
		     ++new_ck, --num_cks) {
			new_ck->type_fgs |= DB_CK_FG_OBS;
		}
	}

	/* If this record pushed us past a threshold for at least one
	 * checksum, then try to generate a summary of our own previously
	 * delayed reports even if this record was not our own. */
	if (summarize
	    && !summarize_rcd(0))
		return 0;

	return 1;
}



/* the database must be locked */
static u_char
add_del(const DCC_CK *del_ck)
{
	DB_RCD del_rcd;

	memset(&del_rcd, 0, sizeof(del_rcd));
	get_ts(&del_rcd.ts);
	DB_TGTS_RCD_SET(&del_rcd, DCC_TGTS_DEL);
	del_rcd.srvr_id_auth = my_srvr_id;
	del_rcd.fgs_num_cks = 1;
	del_rcd.cks[0].type_fgs = del_ck->type;
	memcpy(del_rcd.cks[0].sum, del_ck->sum, sizeof(del_rcd.cks[0].sum));
	if (!db_add_rcd(dcc_emsg, &del_rcd)) {
		DB_ERROR_MSG2("add delete", dcc_emsg);
		return 0;
	}

	return 1;
}



static const DCC_CK *
start_work(QUEUE *q)
{
	const DCC_CK *ck, *ck_lim;
	DCC_CK_TYPES type, prev_type;
	int num_cks;

	num_cks = q->pkt_len - (sizeof(q->pkt.r) - sizeof(q->pkt.r.cks));
	if (num_cks < 0
	    || num_cks > ISZ(q->pkt.r.cks)
	    || num_cks % sizeof(DCC_CK) != 0) {
		forget_error(q, "packet length %d wrong for %s",
			     q->pkt_len, from_id_ip(q, 1));
		return 0;
	}
	num_cks /= sizeof(DCC_CK);

	/* send previous answer if this is a retransmission */
	if (ridc_get(q)) {
		repeat_resp(q);
		return 0;
	}

	if (db_failed_line)		/* be silent while database bad */
		return 0;

	ck = q->pkt.r.cks;
	ck_lim = &q->pkt.r.cks[num_cks];

	/* check each checksum */
	for (prev_type = DCC_CK_INVALID; ck < ck_lim; ++ck, prev_type = type) {
		if (ck->len != sizeof(*ck)) {
			forget_error(q, "unknown checksum length %d%s",
				     ck->len, from_id_ip(q, 0));
			return 0;
		}
		/* requiring that the checksums be ordered makes it easy
		 * to check for duplicates and for bogus long packets */
		type = ck->type;
		if (!DCC_CK_OK_DCC_CLNT(grey_on, type)) {
			forget_error(q, "unknown checksum %s%s",
				     DB_TYPE2STR(type), from_id_ip(q, 0));
			return 0;
		}
		if (prev_type >= type) {
			forget_error(q, "out of order %s checksum%s",
				     DB_TYPE2STR(ck->type), from_id_ip(q, 0));
			return 0;
		}
	}

	if (db_lock() < 0) {
		NORESP_EMSG(q);
		return 0;
	}

	return ck_lim;
}



/* send the response and release q */
static void
fin_work(const QUEUE *q, DCC_HDR *answer)
{
	int delay_us;

	/* send the response */
	answer->op = DCC_OP_ANSWER;
	send_resp(q, answer, 0);

	/* update the average queue delay, unless it is crazy */
	gettimeofday(&db_time, 0);
	delay_us = tv_diff2us(&db_time, &q->answer);
	if (delay_us < 0)
		return;

	update_q_delay();
	q_delays[0].us += delay_us;
	++q_delays[0].ops;
}



/* use only db_sts.hash and db_sts.rcd2
 * release q on failure */
static u_char
make_answer(QUEUE *q,
	    const DCC_CK *ck_lim,
	    u_char have_rcd,		/* db_sts.sumrcd.d.r is new record */
	    DCC_ANSWER *answer,
	    DCC_TGTS gross_tgts,	/* total for this report, maybe MANY */
	    DCC_TGTS* max_tgts)		/* statistics */
{
	const DCC_CK *ck;
	DCC_TGTS c_tgts;		/* current count with this report */
	DCC_TGTS p_tgts;		/* count before this report */
	DCC_ANSWER_BODY_CKS *b;
	DCC_CK_TYPES type;
	const DB_RCD_CK *rcd_ck, *prev_rcd_ck;
	int num_rcd_cks;
	DB_PTR prev;
	*max_tgts = 0;

	if (have_rcd) {
		rcd_ck = db_sts.sumrcd.d.r->cks;
		num_rcd_cks = DB_NUM_CKS(db_sts.sumrcd.d.r);
	} else {
		num_rcd_cks = 0;
		rcd_ck = 0;
	}
	b = answer->b;
	for (ck = q->pkt.r.cks; ck < ck_lim; ++ck) {
		type = ck->type;
		if (num_rcd_cks > 0
		    && type == DB_CK_TYPE(rcd_ck)) {
			/* try to copy answer from report's new record */
			c_tgts = DB_TGTS_CK(rcd_ck);
			if (c_tgts < DCC_TGTS_TOO_MANY) {
				p_tgts = c_tgts - gross_tgts;
			} else if (prev = DB_PTR_EX(rcd_ck->prev),
				   prev == DB_PTR_NULL) {
				p_tgts = 0;
			} else {
				prev_rcd_ck = db_map_rcd_ck(dcc_emsg,
							&db_sts.rcd2,
							prev, type);
				if (!prev_rcd_ck) {
					DB_ERROR_MSG(dcc_emsg);
					RIDC_BAD(q);
					return 0;
				}
				p_tgts = DB_TGTS_CK(prev_rcd_ck);
			}
			--num_rcd_cks;
			++rcd_ck;

		} else {
			if (!get_ck_tgts(&p_tgts, 0, 0, type, ck->sum)) {
				NORESP_EMSG(q);
				return 0;
			}
			if (DB_TEST_NOKEEP(db_parms.nokeep_cks, type)) {
				/* uninteresting checksums have no value
				 * unless they are whitelisted */
				c_tgts = p_tgts;
				if (p_tgts == 0)
					p_tgts = DCC_TGTS_INVALID;
			} else {
				c_tgts = db_sum_ck(p_tgts, gross_tgts, type);
			}
		}

		b->c = htonl(c_tgts);
		b->p = htonl(p_tgts);
#ifdef DCC_PKT_VERSION5
		if (q->pkt.hdr.pkt_vers <= DCC_PKT_VERSION5)
			b = (DCC_ANSWER_BODY_CKS *)&b->p;
		else
#endif
		++b;

		if (*max_tgts < c_tgts
		    && c_tgts <= DCC_TGTS_OK2) {
			*max_tgts = c_tgts;
			/* Complain about failures to whitelist by
			 * trusted clients.  The main use of this is
			 * to detect whitelisting failures of IP addresses
			 * such as 127.0.0.1 for reputations, and those
			 * matter only for known clients.  */
			if ((p_tgts >= DCC_TGTS_OK)
			    && !(q->flags & Q_FG_UNTRUSTED))
				TMSG4(WLIST, "%s whitelisted %s %s%s",
				      qop2str(q),
				      DB_TYPE2STR(type),
				      dcc_ck2str_err(type, ck->sum, 0),
				      from_id_ip(q, 0));
		}
	}
	answer->hdr.len = (sizeof(*answer) - sizeof(answer->b)
			   + ((char *)b - (char *)answer->b));
	return 1;
}



/* release q on failure
 *	the database must be locked */
static u_char
do_report(QUEUE *q,
	  DCC_TGTS tgts0, const DCC_CK *ck_lim,
	  DCC_ANSWER *answer,
	  DCC_TGTS *max_tgts)
{
	const DCC_CK *ck;
	DCC_TGTS tgts;
	DCC_TGTS gross_tgts;		/* DCC_TGTS_TOO_MANY if spam */
	DB_PTR rcd_pos;
	DB_RCD new;
	DB_RCD_CK *new_ck;
	DCC_CK_TYPES type;
	char tgts_buf[DCC_XHDR_MAX_TGTS_LEN];

	tgts = tgts0;
	if (tgts & (DCC_TGTS_SPAM | DCC_TGTS_REP_SPAM)) {
		tgts &= DCC_TGTS_MASK;
		if (tgts == 0)
			tgts = 1;
		gross_tgts = DCC_TGTS_TOO_MANY;
	} else if (tgts == DCC_TGTS_TOO_MANY) {
		tgts = 1;
		gross_tgts = DCC_TGTS_TOO_MANY;
	} else if (tgts > DCC_TGTS_RPT_MAX) {
		forget_error(q, "bogus target count %s%s",
			     dcc_tgts2str(tgts_buf, sizeof(tgts_buf),
					  tgts, grey_on),
			     from_id_ip(q, 0));
		return 0;
	} else {
		gross_tgts = tgts;
	}

	if (gross_tgts < 10) {
		;
	} else if (gross_tgts == DCC_TGTS_TOO_MANY) {
		++dccd_stats.reportmany;
	} else if (gross_tgts > 1000) {
		++dccd_stats.report1000;
	} else if (gross_tgts > 100) {
		++dccd_stats.report100;
	} else if (gross_tgts > 10) {
		++dccd_stats.report10;
	}

	/* Get ready to add the report to the database,
	 * and as a side effect, find the data to answer the query.
	 * Start by creating the record to add to the database. */
	get_ts(&new.ts);
	new.srvr_id_auth = my_srvr_id;
	DB_TGTS_RCD_SET(&new, gross_tgts);

	/* copy checksums to the new record */
	new.fgs_num_cks = 0;
	new_ck = new.cks;
	for (ck = q->pkt.r.cks; ck < ck_lim; ++ck) {
		type = ck->type;
		if (DB_TEST_NOKEEP(db_parms.nokeep_cks, type))
			continue;
		memcpy(new_ck->sum, ck->sum, sizeof(new_ck->sum));
		new_ck->type_fgs = type;
		++new_ck;
		++new.fgs_num_cks;
	}

	if (!(q->flags & Q_FG_RPT_OK)) {
		/* finished if this is a query */
		return make_answer(q, ck_lim, 0, answer, gross_tgts,
				   max_tgts);
	}

	if (new.fgs_num_cks == 0) {
		rcd_pos = DB_PTR_NULL;
	} else {
		/* Add the record to the database.
		 * That will update the totals for each checksum */
		if (!add_dly_rcd(&new, 0)) {
			NORESP_EMSG(q);
			return 0;
		}
		rcd_pos = db_sts.sumrcd.s.rptr;
	}

	/* generate the response, perhaps from the new record */
	return make_answer(q, ck_lim, rcd_pos!=DB_PTR_NULL, answer, gross_tgts,
			   max_tgts);
}



/* process a single real request */
void
do_work(QUEUE *q)
{
	const DCC_CK *ck_lim;
	DCC_ANSWER answer;
	DCC_TGTS max_tgts, tgts;

	ck_lim = start_work(q);
	if (!ck_lim)
		return;

	tgts = 0;
	switch (q->pkt.hdr.op) {
	case DCC_OP_QUERY:
		++dccd_stats.queries;
		q->flags &= ~Q_FG_RPT_OK;
		break;

	case DCC_OP_REPORT:
		if (!(q->flags & Q_FG_RPT_OK)) {
			++dccd_stats.report_reject;
			clnt_msg(q, "treat %s as query", from_id_ip(q, 1));
			++dccd_stats.queries;
		} else {
			tgts = ntohl(q->pkt.r.tgts);
			++dccd_stats.reports;
		}
		break;

	case DCC_OP_INVALID:
	case DCC_OP_NOP:
	case DCC_OP_ANSWER:
	case DCC_OP_ADMN:
	case DCC_OP_OK:
	case DCC_OP_ERROR:
	case DCC_OP_DELETE:
	case DCC_OP_GREY_REPORT:
	case DCC_OP_GREY_QUERY:
	case DCC_OP_GREY_SPAM:
	case DCC_OP_GREY_WHITE:
		dcc_logbad(EX_SOFTWARE, "impossible queued operation");
		break;
	}

	if (!do_report(q, tgts, ck_lim, &answer,
		       &max_tgts)) {
		/* ensure that the clock ticks so rate limits don't stick */
		gettimeofday(&db_time, 0);
	} else {
		/* notice the size of our answer */
		if (max_tgts == DCC_TGTS_OK || max_tgts == DCC_TGTS_OK2) {
			++dccd_stats.respwhite;
		} else if (max_tgts == DCC_TGTS_TOO_MANY) {
			++dccd_stats.respmany;
		} else if (max_tgts > 1000) {
			++dccd_stats.resp1000;
		} else if (max_tgts > 100) {
			++dccd_stats.resp100;
		} else if (max_tgts > 10) {
			++dccd_stats.resp10;
		}

		fin_work(q, &answer.hdr);
	}
}



/* return 0 for a new embargo,
 *	embargo count	for an existing embargo,
 *	DCC_TGTS_TOO_MANY   no embargo
 *	DCC_TGTS_OK	    a newly expired embargo
 *	DCC_TGTS_INVALID    broken database */
static DCC_TGTS
search_grey(const DCC_CK *req_ck3,	/* triple checksum */
	    const DCC_CK *req_ckb,	/* body seen with it */
	    u_char body_known)
{
	DB_RCD_CK *ck3, *ckb;
	DB_PTR prev3;
	DCC_TS old_ts;
	DCC_TGTS result_tgts;
	int i;

	/* look for the triple checksum */
	switch (db_lookup(dcc_emsg, DCC_CK_GREY3, req_ck3->sum,
			  0, MAX_HASH_ENTRIES,
			  &db_sts.hash, &db_sts.rcd, &ck3)) {
	case DB_FOUND_EMPTY:
	case DB_FOUND_CHAIN:
	case DB_FOUND_INTRUDER:
		return 0;

	case DB_FOUND_IT:
		/* We found the triple checksum.
		 * If it is marked ok (MANY) or deleted,
		 * then we have our answer */
		result_tgts = DB_TGTS_CK(ck3);
		if (result_tgts == DCC_TGTS_TOO_MANY || result_tgts == 0)
			return result_tgts;

		/* Otherwise look for a report of the triple with
		 * the right body checksum that is old enough. */
		result_tgts = 0;
		dcc_timeval2ts(&old_ts, &db_time, -grey_embargo);
		for (;;) {
			ckb = db_sts.rcd.d.r->cks;
			for (i = DB_NUM_CKS(db_sts.rcd.d.r);
			     i > 0;
			     --i, ++ckb) {
				/* try the next report in the database
				 * if it has the wrong body checksum
				 *
				 * If we are weak on bodies,
				 * act as if all reports of the triple checksums
				 * are with the right body checksum. */
				if (!grey_weak_body && req_ckb) {
					if (DB_CK_TYPE(ckb) != DCC_CK_BODY)
					    continue;
					if (memcmp(req_ckb->sum, ckb->sum,
						   sizeof(DCC_SUM)))
					    break;
				}

				/* We found the right body checksum in
				 * chain of the triple checksum
				 * or we don't care.
				 *
				 * If the report is old enough, then
				 * the embargo is over. */
				if (dcc_ts_newer_ts(&old_ts,
						    &db_sts.rcd.d.r->ts))
					return DCC_TGTS_OK;

				/* If it is not old enough,
				 * then we know this is not a new embargo for
				 * this body (i.e. the reported target count
				 * will be >0) and we must keep looking for an
				 * old enough report with the body checksum. */
				++result_tgts;
				break;
			}

			/* If we know the body checksum is not in the database,
			 * then there is no profit in looking at other reports
			 * of the triple checksum to try to find an old enough
			 * report that is with the right body checksum.
			 * We know this is a new embargo. */
			if (!body_known)
				return 0;

			/* If we reach the end of the chain of the
			 * triple checksum without finding an old
			 * enough report for the right body,
			 * then the embargo is not over. */
			prev3 = DB_PTR_EX(ck3->prev);
			if (prev3 == DB_PTR_NULL)
				return result_tgts;

			/* examine the timestamp of the preceding report
			 * of the triple */
			ck3 = db_map_rcd_ck(dcc_emsg, &db_sts.rcd,
					    prev3, DCC_CK_GREY3);
			if (!ck3)
				return DCC_TGTS_INVALID;
		}
		break;

	case DB_FOUND_LATER:
	case DB_FOUND_SYSERR:
		DB_ERROR_MSG(dcc_emsg);
		return DCC_TGTS_INVALID;
	}
	return DCC_TGTS_INVALID;
}



void
do_grey(QUEUE *q)
{
	DCC_OPS op;
	DB_RCD new;
	const DCC_CK *req, *req_lim;
	const DCC_CK *req_ck_ip, *req_ck_triple, *req_ck_msg, *req_ck_body;
	u_char body_known;
	DB_RCD_CK *new_ck, *found_ck;
	DCC_GREY_ANSWER resp;
	DCC_TGTS tgts;
	DCC_TGTS ip_tgts;		/* existing count for DCC_CK_IP */
	DCC_TGTS triple_tgts;		/*   "   count for GREY_TRIPLE */
	DCC_TGTS msg_tgts;		/*   "   count for GREY_MSG */
	DCC_TGTS eff_msg_tgts;		/* effective value: 0=reported to DCC */
	DCC_TGTS new_msg_tgts;		/* value after this */
	DCC_TGTS result_tgts;		/* no embargo, ending, whitelist or # */

	TMSG1(QUERY, "received %s", op_id_ip(q));
	if (!ck_clnt_id(q))
		return;
	if (q->flags & Q_FG_UNTRUSTED) {
		anon_msg("drop %s", from_id_ip(q, 1));
		return;
	}

	/* an embargo of 0 seconds means we should only collect names */
	op = q->pkt.hdr.op;
	if (op == DCC_OP_GREY_REPORT && grey_embargo == 0)
		op = DCC_OP_GREY_WHITE;

	req_lim = start_work(q);
	if (!req_lim)
		return;

	/* Require
	 *	the body checksum,
	 *	the checksum of the (body,sender,target),
	 *	and the checksum of the (source,sender,target) triple.
	 * Allow other checksums for whitelisting. */
	ip_tgts = 0;
	body_known = grey_weak_body;
	req_ck_ip = 0;
	req_ck_body = 0;
	req_ck_triple = 0;
	req_ck_msg = 0;
	msg_tgts = eff_msg_tgts = 0;
	for (req = q->pkt.r.cks; req < req_lim; ++req) {
		/* Note our main checksums of the greylist triple and
		 * the message body.  Search the database for it later */
		if (DCC_CK_IS_GREY_TRIPLE(1, req->type)) {
			req_ck_triple = req;
			continue;
		}

		if (!DCC_CK_OK_GREY_CLNT(req->type))
			continue;	/* ignore unknown checksums */
		switch (req->type) {
		case DCC_CK_IP:
			req_ck_ip = req;
			break;
		case DCC_CK_BODY:
			req_ck_body = req;
			break;
		case DCC_CK_GREY_MSG:
			req_ck_msg = req;
			break;
		}
		/* check for whitelisting and whether this is a new embargo */
		switch (db_lookup(dcc_emsg, req->type, req->sum,
				  0, MAX_HASH_ENTRIES,
				  &db_sts.hash, &db_sts.rcd, &found_ck)) {
		case DB_FOUND_LATER:
		case DB_FOUND_SYSERR:
			DB_ERROR_MSG(dcc_emsg);
			RIDC_BAD(q);
			return;
		case DB_FOUND_IT:
			/* ignore deleted checksums */
			tgts = DB_TGTS_CK(found_ck);
			if (tgts == 0)
				continue;

			/* honor whitelisting */
			if (tgts == DCC_TGTS_GREY_WHITE
			    && op != DCC_OP_GREY_WHITE) {
				op = DCC_OP_GREY_WHITE;
				++dccd_stats.respwhite;
			}

			switch (req->type) {
			case DCC_CK_BODY:
				/* notice if the target body exists at all */
				body_known = 1;
				break;
			case DCC_CK_GREY_MSG:
				msg_tgts = tgts;
				if (msg_tgts != DCC_TGTS_TOO_MANY) {
					/* this is an old embargo that has
					 * already been reported by the client
					 * to a normal DCC server */
					eff_msg_tgts = 1;
				}
				break;
			case DCC_CK_IP:
				ip_tgts = tgts;
				break;
			default:
				break;
			}
			break;
		case DB_FOUND_EMPTY:
		case DB_FOUND_CHAIN:
		case DB_FOUND_INTRUDER:
			break;
		}
	}
	if (!req_ck_triple) {
		send_error(q, "missing %s checksum for %s",
			   DB_TYPE2STR(DCC_CK_GREY3), qop2str(q));
		return;
	}
	if (op == DCC_OP_GREY_REPORT && !grey_weak_body) {
		if (!req_ck_body) {
			send_error(q, "missing body checksum for %s",
				   qop2str(q));
			return;
		}
		if (!req_ck_msg) {
			send_error(q, "missing %s checksum for %s",
				   DB_TYPE2STR(DCC_CK_GREY_MSG), qop2str(q));
			return;
		}
	}

	/* decide if the embargo should end */
	triple_tgts = search_grey(req_ck_triple, req_ck_body, body_known);
	if (triple_tgts == DCC_TGTS_INVALID) {
		NORESP_EMSG(q);		/* broken database */
		return;
	}
	/* End existing embargo on a newly whitelisted sender so its
	 *	messages are logged.
	 * Quietly prevent future embargos of whitelisted senders that have
	 *	not been greylisted.
	 * Honor grey_weak_ip whitelisting even after it is turned off */
	if (triple_tgts >= DCC_TGTS_TOO_MANY) {
		result_tgts = triple_tgts;
	} else if (op == DCC_OP_GREY_WHITE) {
		result_tgts = eff_msg_tgts ? DCC_TGTS_OK : DCC_TGTS_TOO_MANY;
	} else if (ip_tgts == DCC_TGTS_TOO_MANY) {
		result_tgts = DCC_TGTS_TOO_MANY;
	} else {
		result_tgts = triple_tgts;
	}

	if (op == DCC_OP_GREY_QUERY) {
		++dccd_stats.queries;

	} else if (!(q->flags & Q_FG_RPT_OK)) {
		++dccd_stats.report_reject;
		clnt_msg(q, "treat %s as query", from_id_ip(q, 1));
		++dccd_stats.queries;

	} else {
		/* add a report for this message */
		++dccd_stats.reports;
		new.srvr_id_auth = my_srvr_id;
		new_ck = new.cks;
		new.fgs_num_cks = 0;
		if (result_tgts < DCC_TGTS_TOO_MANY) {
			if (req_ck_body) {
				new_ck->type_fgs = DCC_CK_BODY;
				memcpy(new_ck->sum, req_ck_body->sum,
				       sizeof(new_ck->sum));
				++new.fgs_num_cks;
				++new_ck;
			}
			new_msg_tgts = 1;
			DB_TGTS_RCD_SET(&new, 1);
		} else {
			/* embargo now ending (DCC_TGTS_TOO_OK)
			 * or no embargo (DCC_TGTS_TOO_MANY) */
			if (grey_weak_ip && req_ck_ip) {
				new_ck->type_fgs = DCC_CK_IP;
				memcpy(new_ck->sum, req_ck_ip->sum,
				       sizeof(new_ck->sum));
				++new.fgs_num_cks;
				++new_ck;
			}
			new_msg_tgts = 0;
			DB_TGTS_RCD_SET(&new, DCC_TGTS_TOO_MANY);
		}

		/* Include the GREY_MSG checksum in the database
		 * record for a new embargo.
		 * The message checksum lets an SMTP server report an
		 * embargoed message to the DCC before the embargo is over,
		 * but not report it more than once even if more than one
		 * SMTP client retransmits the message.
		 *
		 * If the GREY_MSG checksum does not exist in the
		 * database, then tell the DCC client the message is new
		 * and should be reported to the DCC server.  We must put the
		 * the _GREY_MSG into the database so we will recognize
		 * the message as not new when it is retransmitted.
		 *
		 * If the GREY_MSG checksum exists and is not MANY,
		 * then we may have a retransmission of the message
		 * from another IP address.
		 * We need to tell the DCC client to not report to the
		 * DCC server. The new value for the CK_GREY_MSG checksum
		 * should be whatever we are using for the triple checksum.
		 *
		 * If the existing count for the GREY_MSG checksum is
		 * MANY, and the new value for triple checksum is not MANY,
		 * then we have a new copy of the message and a new embargo.
		 * We have a spammer with multiple senders instead of a
		 * legitimate multihomed SMTP client.  We need to tell the
		 * DCC client to report to the DCC server.  To remember
		 * that we told the DCC client to report to the DCC server,
		 * we must first delete the existing MANY report of the
		 * GREY_MSG checksum. */
		if (eff_msg_tgts != new_msg_tgts
		    && req_ck_msg) {
			if (msg_tgts == DCC_TGTS_TOO_MANY
			    && !add_del(req_ck_msg)) {
				NORESP_EMSG(q);
				return;
			}
			new_ck->type_fgs = DCC_CK_GREY_MSG;
			memcpy(new_ck->sum, req_ck_msg->sum,
			       sizeof(new_ck->sum));
			++new.fgs_num_cks;
			++new_ck;
		}

		/* Add the triple checksum if we are not whitelisting
		 *	by the IP address
		 * or triple checksum is not new.
		 * We do not want to leave any dangling triples in the
		 * database */
		if (!(grey_weak_ip && req_ck_ip)
		    || result_tgts != DCC_TGTS_TOO_MANY) {
			new_ck->type_fgs = DCC_CK_GREY3;
			memcpy(new_ck->sum, req_ck_triple->sum,
			       sizeof(new_ck->sum));
			++new.fgs_num_cks;
		}

		get_ts(&new.ts);
		if (!db_add_rcd(dcc_emsg, &new)) {
			DB_ERROR_MSG(dcc_emsg);
			RIDC_BAD(q);
			return;
		}
	}

	/* In the result sent to the DCC client,
	 * the triple checksum is preceeded by the message checksum
	 * with a count of 0 if this is a new embargo.
	 * Targets of messages of new embargos should be counted among
	 * total targets in reports sent to DCC servers.  After they
	 * have been included in such an early report to a DCC server,
	 * they should never be included again, except for bad reputations. */
	resp.msg = htonl(eff_msg_tgts);

	/* Answer SMTP DATA command greylist operations with the target
	 * count of the triple checksum:
	 *	DCC_TGTS_OK if the embargo is just now being removed
	 *	DCC_TGTS_TOO_MANY if there is no current embargo
	 *	DCC_TGTS_GREY_WHITE if whitelisted.
	 *	embargo # otherwise */
	resp.triple = htonl(result_tgts);
	resp.hdr.len = sizeof(resp);

	fin_work(q, &resp.hdr);
}



static time_t
picky_time(const QUEUE *q)
{
	time_t ts, delta;

	/* If the request arrived while we were asleep, then the client's
	 * timestamp ought to be smaller than when select() finished and
	 * we think the request arrived. */
	ts = ntohl(q->pkt.d.date);
	delta = ts - q->answer.tv_sec;
	if (delta <= 0)
		return delta;

	/* If the request arrived while we were handling some other request,
	 * then its timestamp can be larger than the select() wake-up time
	 * but should not be in the future. */
	delta = ts - db_time.tv_sec;
	if (delta < 0)
		delta = 0;
	return delta;
}



static u_char				/* 0=refuse the bad guy, 1=continue */
picky_admn(const QUEUE *q, u_char any_id, u_char any_time)
{
	time_t delta;

	if ((q->flags & Q_FG_UNTRUSTED)
	    || (q->clnt_id != my_srvr_id && !any_id)) {
		forget_error(q, "drop %s", from_id_ip(q, 1));
		return 0;
	}

	if (any_id && any_time)
		return 1;

	/* Demand a current timestamp to guard against replay attacks.
	 * This requires that administrators have clocks close to servers',
	 * and that network and server delays be reasonable. */
	delta = picky_time(q);
	if (delta < -MAX_CMD_CLOCK_SKEW || delta > MAX_CMD_CLOCK_SKEW) {
		send_error(q, "drop %s; timestamp off by %d seconds",
			   qop2str(q), (int)delta);
		return 0;
	}

	return 1;
}



/* the database must be locked */
static u_char				/* 1=ok, 0=error sent to client */
delete_sub(QUEUE *q, DCC_CK *del_ck,
	   u_char grey_spam)
{
	DB_RCD_CK *rcd_ck;
	char buf[80];
	DB_PTR prev;
	DCC_TGTS tgts;

	buf[0] = '\0';
	switch (db_lookup(dcc_emsg, del_ck->type, del_ck->sum,
			  0, MAX_HASH_ENTRIES,
			  &db_sts.hash, &db_sts.rcd, &rcd_ck)) {
	case DB_FOUND_EMPTY:
	case DB_FOUND_CHAIN:
	case DB_FOUND_INTRUDER:
		/* finished if we have not greylisted the spammer */
		if (grey_spam)
			return 1;

		/* ordinary deletions need a delete request added
		 * to the database and flooded */
		snprintf(buf, sizeof(buf), "\"%s %s\" not found to delete",
			 DB_TYPE2STR(del_ck->type),
			 dcc_ck2str_err(del_ck->type, del_ck->sum, 0));

		if (del_ck->type == DCC_CK_SRVR_ID) {
			send_error(q, "%s", buf);
			return 0;
		}
		break;

	case DB_FOUND_IT:
		tgts = DB_TGTS_CK(rcd_ck);
		/* handle an ordinary delete request */
		if (!grey_spam) {
			if (tgts == 0)
				snprintf(buf, sizeof(buf),
					 "%s %s already deleted",
					 DB_TYPE2STR(del_ck->type),
					 dcc_ck2str_err(del_ck->type,
							del_ck->sum, 0));
			break;
		}
		/* We are deleting a greylist checksum.
		 * If we are deleting very new greylist records,
		 * we can cheat and avoid adding to the database
		 * by scribbling over the records.
		 * If there is an older record that might have been flooded,
		 * we must add a delete request to the database
		 * that will itself be flooded. */
		for (;;) {
			/* finished if the target has already been deleted */
			if (tgts == 0)
				return 1;
			if (db_sts.rcd.s.rptr < oflods_max_cur_pos
			    || oflods_max_cur_pos == 0) {
				/* We need to add a delete request, because
				 * the record might have been flooded */
				break;
			}
			prev = DB_PTR_EX(rcd_ck->prev);
			/* try to delete the entire greylist entry
			 * starting with the target triple checksum */
			do {
				/* only if the embargo is not over */
				if (DB_TGTS_CK(rcd_ck) >= DCC_TGTS_TOO_MANY)
					goto need_rcd;
				DB_TGTS_CK_SET(rcd_ck, 0);
			} while (--rcd_ck >= db_sts.rcd.d.r->cks);
			DB_TGTS_RCD_SET(db_sts.rcd.d.r, 0);
			SET_FLUSH_RCD_HDR(&db_sts.rcd, 1);

			/* stop after the last record */
			if (prev == DB_PTR_NULL)
				return 1;

			rcd_ck = db_map_rcd_ck(dcc_emsg, &db_sts.rcd,
					       prev, del_ck->type);
			if (!rcd_ck) {
				NORESP_EMSG(q);
				return 0;
			}
			tgts = DB_TGTS_CK(rcd_ck);
		}
need_rcd:;
		break;

	case DB_FOUND_LATER:
	case DB_FOUND_SYSERR:
		DB_ERROR_MSG(dcc_emsg);
		RIDC_BAD(q);
		return 0;
	}

	/* Add the delete request to the database even if the
	 * checksum seems deleted or absent so that we will
	 * flood the delete request.  This is required to ensure that
	 * records get deleted when they are created at one DCC server
	 * and deleted at another. */
	if (!add_del(del_ck))
		BUFCPY(buf, dcc_emsg);

	if (buf[0] != '\0') {
		send_error(q, "%s", buf);
		return 0;
	}

	TMSG3(ADMN, "deleted %s %s%s",
	      DB_TYPE2STR(del_ck->type),
	      dcc_ck2str_err(del_ck->type, del_ck->sum, 0),
	      from_id_ip(q, 0));
	return 1;
}



void
do_delete(QUEUE *q)
{
	if (!ck_clnt_srvr_id(q))
		return;
	if (!picky_admn(q, 0, 0))
		return;
	/* if we've already answered, then just repeat ourselves */
	if (ridc_get(q)) {
		repeat_resp(q);
		return;
	}

	dcc_error_msg("received %s", op_id_ip(q));
	++dccd_stats.admin;

	if (q->pkt_len != sizeof(q->pkt.d)) {
		send_error(q, "wrong packet length %d for %s",
			   q->pkt_len, qop2str(q));
		return;
	}
	if (q->pkt.d.ck.len != sizeof(q->pkt.d.ck)) {
		send_error(q, "unknown checksum length %d", q->pkt.d.ck.len);
		return;
	}
	if (!DCC_CK_OK_DB(grey_on, q->pkt.d.ck.type)) {
		send_error(q, "unknown checkksum type %d", q->pkt.d.ck.type);
		return;
	}

	if (db_lock() < 0) {
		NORESP_EMSG(q);
		return;
	}
	if (delete_sub(q, &q->pkt.d.ck, 0)) {
		/* We need to clean the database after a deletion
		 * to correct the totals of other checksums.
		 * Don't bother for reputations or server-ID declarations. */
		if (!DCC_CK_IS_REP_CMN(grey_on, q->pkt.d.ck.type)
		    && q->pkt.d.ck.type != DCC_CK_SRVR_ID)
			need_del_dbclean = "checksum deleted";

		send_ok(q);
	}
}



/* restore the embargo against a sender of spam */
void
do_grey_spam(QUEUE *q)
{
	TMSG1(QUERY, "received %s", op_id_ip(q));
	if (!ck_clnt_id(q))
		return;
	if (q->flags & Q_FG_UNTRUSTED) {
		anon_msg("drop %s", from_id_ip(q, 1));
		return;
	}

	/* require the checksum of the (source,sender,target) triple */
	if (q->pkt_len != sizeof(q->pkt.gs)) {
		send_error(q, "wrong packet length %d for %s",
			   q->pkt_len, qop2str(q));
		return;
	}
	if (q->pkt.gs.triple.type != DCC_CK_GREY3) {
		send_error(q, "%s instead of %s for %s",
			   DB_TYPE2STR(q->pkt.gs.msg.type),
			   DB_TYPE2STR(DCC_CK_GREY3),
			   qop2str(q));
		return;
	}
	if (q->pkt.gs.triple.len != sizeof(q->pkt.gs.triple)) {
		send_error(q, "unknown triple checksum length %d",
			   q->pkt.gs.ip.len);
		return;
	}
	if (q->pkt.gs.msg.type != DCC_CK_GREY_MSG) {
		send_error(q, "%s instead of %s for %s",
			   DB_TYPE2STR(q->pkt.gs.msg.type),
			   DB_TYPE2STR(DCC_CK_GREY_MSG),
			   qop2str(q));
		return;
	}
	if (q->pkt.gs.msg.len != sizeof(q->pkt.gs.msg)) {
		send_error(q, "unknown msg checksum length %d",
			   q->pkt.gs.ip.len);
		return;
	}
	if (q->pkt.gs.ip.type != DCC_CK_IP) {
		send_error(q, "%s instead of %s for %s",
			   DB_TYPE2STR(q->pkt.gs.msg.type),
			   DB_TYPE2STR(DCC_CK_IP),
			   qop2str(q));
		return;
	}
	if (q->pkt.gs.ip.len != sizeof(q->pkt.gs.ip)) {
		send_error(q, "unknown IP checksum length %d",
			   q->pkt.gs.ip.len);
		return;
	}

	if (db_lock() < 0) {
		NORESP_EMSG(q);
		return;
	}
	if (delete_sub(q, &q->pkt.gs.ip, 1)
	    && delete_sub(q, &q->pkt.gs.triple, 1)
	    && delete_sub(q, &q->pkt.gs.msg, 1))
		send_ok(q);
}



static void
do_flod(QUEUE *q)
{
	DCC_ADMN_RESP check;
	int print_len;
	u_int32_t val, arg;
	DCC_AOP_FLODS fop;
	FLOD_MMAP *mp;
	OFLOD_INFO *ofp;
	u_char loaded, found_it;

	val = ntohl(q->pkt.ad.val1);
	fop = val % 256;
	arg = val / 256;

	if (fop != DCC_AOP_FLOD_LIST) {
	    if (!picky_admn(q, fop == DCC_AOP_FLOD_STATS, 0))
		    return;
	}

	switch (fop) {
	case DCC_AOP_FLOD_CHECK:
		/* `cdcc "flood check"` forces occasional defenses of
		 * our server-ID */
		if (host_id_next > db_time.tv_sec + 60)
			host_id_next = db_time.tv_sec;

		next_flods_ck = 0;
		if (0 >= check_load_ids(0)) {
			dcc_error_msg("%s", dcc_emsg);
			send_error(q, "%s", dcc_emsg);
			return;
		}
		flod_stats_printf(check.val.string, sizeof(check.val.string),
				  (!FLODS_OK() || flods_st == FLODS_ST_OFF)
				  ? 0
				  : (flods_st != FLODS_ST_ON) ? 1
				  : 2,
				  oflods.total, oflods.open, iflods.open);
		check.hdr.len = (strlen(check.val.string)
				 + sizeof(check)-sizeof(check.val));
		check.hdr.op = DCC_OP_ADMN;
		send_resp(q, &check.hdr, 0);
		flods_ck(1);
		check_blacklist_file();
		return;

	case DCC_AOP_FLOD_SHUTDOWN:
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		++flods_off;
		flods_stop("shutdown flooding", 0);
		send_ok(q);
		return;

	case DCC_AOP_FLOD_HALT:
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		++flods_off;
		flods_stop("stop flooding", 1);
		send_ok(q);
		return;

	case DCC_AOP_FLOD_RESUME:
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (0 >= check_load_ids(0)) {
			dcc_error_msg("%s", dcc_emsg);
			send_error(q, "%s", dcc_emsg);
			return;
		}
		if (flods_off) {
			flods_off = 0;
			flods_restart("resume flooding", 0);
		}
		send_ok(q);
		flods_ck(0);
		return;

	case DCC_AOP_FLOD_REWIND:
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (flod_mmaps) {
			loaded = 0;
		} else if (!load_flod(0)) {
			send_error(q, "too busy to rewind floods");
			return;
		} else {
			loaded = 1;
		}
		found_it = (arg == DCC_ID_INVALID);
		for (mp = flod_mmaps->mmaps;
		     mp <= LAST(flod_mmaps->mmaps);
		     ++mp) {
			if (arg == DCC_ID_INVALID
			    || mp->rem_id == arg) {
				mp->flags |= FLODMAP_FG_NEED_REWIND;
				mp->flags &= ~FLODMAP_FG_FFWD_IN;
				dcc_trace_msg("rewind flood from server-ID %d",
					      arg);
				found_it = 1;
			}
		}
		if (!found_it) {
			send_error(q, "unknown server-ID %d for %s",
				   arg, qop2str(q));
		} else {
			send_ok(q);
			flods_ck(0);
		}
		if (loaded)
			oflods_clear();
		return;

	case DCC_AOP_FLOD_LIST:
		loaded = !flod_mmaps && load_flod(0);
		if (flod_mmaps) {
			print_len = flods_list(check.val.string,
					       sizeof(check.val.string),
					       (q->flags & Q_FG_UNTRUSTED)!=0);
		} else {
			/* it is not an error if map is locked, because
			 * dbclean uses this operation to see if we are
			 * listening */
			print_len = snprintf(check.val.string,
					     ISZ(check.val.string),
					     "too busy to list floods");
			if (print_len > ISZ(check.val.string))
				print_len = ISZ(check.val.string);
		}
		check.hdr.len = (print_len
				 + sizeof(check)-sizeof(check.val));
		check.hdr.op = DCC_OP_ADMN;
		send_resp(q, &check.hdr, 0);
		if (loaded)
			oflods_clear();
		return;

	case DCC_AOP_FLOD_STATS:
	case DCC_AOP_FLOD_STATS_CLEAR:
		print_len = flod_stats(check.val.string,
				       sizeof(check.val.string),
				       arg,
				       fop == DCC_AOP_FLOD_STATS_CLEAR);
		if (print_len < 0) {
			send_error(q, "too busy to find flood stats");
			return;
		}
		check.hdr.len = print_len + sizeof(check)-sizeof(check.val);
		check.hdr.op = DCC_OP_ADMN;
		send_resp(q, &check.hdr, 0);
		flods_ck(0);
		return;

	case DCC_AOP_FLOD_FFWD_IN:
	case DCC_AOP_FLOD_FFWD_OUT:
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (flod_mmaps) {
			loaded = 0;
		} else if (!load_flod(0)) {
			send_error(q, "too busy to fast-forward floods");
			return;
		} else {
			loaded = 1;
		}
		ofp = oflods.infos;
		for (;;) {
			mp = ofp->mp;
			if (mp->rem_id == arg) {
				/* found the target */
				if (fop == DCC_AOP_FLOD_FFWD_OUT) {
					ofp->cur_pos = db_csize;
					if (ofp->soc < 0)
					    mp->confirm_pos = db_csize;
					dcc_trace_msg("fast forward flood to"
						      " server-ID %d",
						      arg);
				} else {
					mp->flags |= FLODMAP_FG_FFWD_IN;
					mp->flags &= ~FLODMAP_FG_NEED_REWIND;
				}
				send_ok(q);
				if (!loaded)
					flods_ck(0);
				break;
			}
			if (++ofp > LAST(oflods.infos)) {
				send_error(q, "unknown server-ID %d for %s",
					   arg, qop2str(q));
				break;
			}
		}
		if (loaded)
			oflods_clear();
		return;
	}

	send_error(q, "unrecognized %s value %d", qop2str(q), fop);
}



void
stats_clear(void)
{
	OFLOD_INFO *ofp;

	memset(&dccd_stats, 0, sizeof(dccd_stats));
	for (ofp = oflods.infos; ofp <= LAST(oflods.infos); ++ofp) {
		if (ofp->rem_hostname[0] == '\0')
			continue;

		/* The counts reported to `cdcc stats` are sums
		 * of the dccd_stats and ofp->cnts values.  Bias
		 * the dccd_stats values by the current ofp->cnts values
		 * so the reported counts will be zero.  When the flooding
		 * connection is closed, the ofp->cnts values will be added
		 * to the dccd_stats values. */
		dccd_stats.iflod_total -= ofp->cnts.total;
		dccd_stats.iflod_accepted -= ofp->cnts.accepted;
		dccd_stats.iflod_stale -= ofp->lc.stale.cur;
		dccd_stats.iflod_dup -= ofp->lc.dup.cur;
		dccd_stats.iflod_wlist -= ofp->lc.wlist.cur;
		dccd_stats.iflod_not_deleted -= ofp->lc.not_deleted.cur;
	}

	q_delays_start = 0;

	memset(&db_stats, 0, sizeof(db_stats));
	dccd_stats.reset = db_time;
}



static u_char				/* 1=sent 0=something wrong */
stats_send(QUEUE *q)
{
	DCC_ADMN_RESP stats;
	char tbuf[80];
	OFLOD_INFO *ofp;
	IFLOD_INFO *ifp;
	int oflods_connecting, iflods_connecting;
	SCNTR iflod_total, iflod_accepted, iflod_stale;
	SCNTR iflod_dup, iflod_wlist, iflod_not_deleted;
	char flod_buf[60];
	char clients_reset[40], reset_buf[36], now_buf[20];
	int clients;
	int age;
	const char *client_ovf;
	int blen, plen, len;

	tbuf[0] = '\0';
	if (dccd_tracemask & DCC_TRACE_ADMN_BIT)
		strcat(tbuf, "ADMN ");
	if (dccd_tracemask & DCC_TRACE_ANON_BIT)
		strcat(tbuf, "ANON ");
	if (dccd_tracemask & DCC_TRACE_CLNT_BIT)
		strcat(tbuf, "CLNT ");
	if (dccd_tracemask & DCC_TRACE_RLIM_BIT)
		strcat(tbuf, "RLIM ");
	if (dccd_tracemask & DCC_TRACE_QUERY_BIT)
		strcat(tbuf, "QUERY ");
	if (dccd_tracemask & DCC_TRACE_RIDC_BIT)
		strcat(tbuf, "RIDC ");
	if (dccd_tracemask & DCC_TRACE_FLOD_BIT)
		strcat(tbuf, "FLOOD ");
	if (dccd_tracemask & DCC_TRACE_FLOD2_BIT)
		strcat(tbuf, "FLOOD2 ");
	if (dccd_tracemask & DCC_TRACE_IDS_BIT)
		strcat(tbuf, "IDS ");
	if (dccd_tracemask & DCC_TRACE_BL_BIT)
		strcat(tbuf, "BL ");
	if (dccd_tracemask & DCC_TRACE_DB_BIT)
		strcat(tbuf, "DB ");
	if (dccd_tracemask & DCC_TRACE_WLIST_BIT)
		strcat(tbuf, "WLIST ");

	clients = clients_get(0, 0, 0, 0, 0, 0, 0);
	if (clients >= 0) {
		client_ovf = "";
	} else {
		client_ovf = ">";
		clients = -clients;
	}
	age = db_time.tv_sec - clients_cleared;
	if (age <= 24*60*60) {
		dcc_time2str(clients_reset, sizeof(clients_reset),
			     "since %X", clients_cleared);
	} else if (age <= 3*24*60*60) {
		snprintf(clients_reset, sizeof(clients_reset),
			 "in %d hours", (age + 60*60/2) / (60*60));
	} else {
		snprintf(clients_reset, sizeof(clients_reset),
			 "in %d days", (age + 24*60*60/2) / (24*60*60));
	}

	oflods_connecting = 0;
	iflod_total = dccd_stats.iflod_total;
	iflod_accepted = dccd_stats.iflod_accepted;
	iflod_stale = dccd_stats.iflod_stale;
	iflod_dup = dccd_stats.iflod_dup;
	iflod_wlist = dccd_stats.iflod_wlist;
	iflod_not_deleted = dccd_stats.iflod_not_deleted;
	for (ofp = oflods.infos; ofp <= LAST(oflods.infos); ++ofp) {
		if (ofp->soc >= 0 && !(ofp->flags & OFLOD_FG_CONNECTED))
			++oflods_connecting;
		iflod_total += ofp->cnts.total;
		iflod_accepted += ofp->cnts.accepted;
		iflod_stale += ofp->lc.stale.cur;
		iflod_dup += ofp->lc.dup.cur;
		iflod_wlist += ofp->lc.wlist.cur;
		iflod_not_deleted += ofp->lc.not_deleted.cur;
	}
	iflods_connecting = 0;
	for (ifp = iflods.infos; ifp <= LAST(iflods.infos); ++ifp) {
		if (ifp->soc >= 0 && !(ifp->flags & IFLOD_FG_VERS_CK))
			++iflods_connecting;
	}
	dcc_time2str(reset_buf, sizeof(reset_buf),"%b %d %X",
		     dccd_stats.reset.tv_sec);
	dcc_time2str(now_buf, sizeof(now_buf), "%b %d %X %Z",
		     db_time.tv_sec);

	blen = min(sizeof(stats.val.string), ntohl(q->pkt.ad.val1));
	plen = snprintf(stats.val.string, blen,
	    "    version "DCC_VERSION"  %s%s%stracing %s\n"
	    "%7d hash entries %6d used "L_DWPAT(9)" DB bytes\n"
	    "%5d ms delay  "L_DPAT" NOPs  "L_DPAT""
	    " ADMN  "L_DPAT" query  %s%d clients %s\n",

	    db_minimum_map ? "DB UNLOCKED  " : "",
	    query_only ? "Q-mode  " : "",
	    grey_on ? "greylist  " : "",
	    tbuf[0] ? tbuf : "nothing",

	    HADDR2LEN(db_hash_len), HADDR2LEN(db_hash_used), db_csize,

	    avg_q_delay_ms(q),

	    dccd_stats.nops, dccd_stats.admin, dccd_stats.queries,
	    client_ovf, clients, clients_reset);
	if (plen >= blen)
		plen = blen-1;
	blen -= plen;

	if (grey_on) {
		len = snprintf(&stats.val.string[plen], blen,
	    L_DWPAT(7)" reports "L_DWPAT(2)" whitelisted\n",

	    dccd_stats.reports,
	    dccd_stats.respwhite);

	} else {
		len = snprintf(&stats.val.string[plen], blen,
			       L_DWPAT(8)" reports "
			       L_DWPAT(7)">10 "
			       L_DWPAT(7)">100 "
			       L_DWPAT(7)">1000 "
			       L_DWPAT(7)" many\n"
			       "         answers "L_DWPAT(7)">10 "
			       L_DWPAT(7)">100 "
			       L_DWPAT(7)">1000 "
			       L_DWPAT(7)" many\n",

	    dccd_stats.reports,
	    (dccd_stats.report10 + dccd_stats.report100
	     + dccd_stats.report1000 + dccd_stats.reportmany),
	    (dccd_stats.report100 + dccd_stats.report1000
	     + dccd_stats.reportmany),
	    dccd_stats.report1000 + dccd_stats.reportmany,
	    dccd_stats.reportmany,

	    (dccd_stats.resp10 + dccd_stats.resp100
	     + dccd_stats.resp1000 + dccd_stats.respmany),
	    dccd_stats.resp100 + dccd_stats.resp1000 + dccd_stats.respmany,
	    dccd_stats.resp1000 + dccd_stats.respmany,
	    dccd_stats.respmany);
	}
	if (len >= blen)
		len = blen-1;
	blen -= len;
	plen += len;

	len = snprintf(&stats.val.string[plen], blen,
	    L_DWPAT(8)" bad op "
	    L_DWPAT(4)" passwd "
	    L_DWPAT(6)" blist "
	    L_DWPAT(4)" reject "
	    L_DWPAT(6)" retrans\n",
	    dccd_stats.bad_op, dccd_stats.bad_passwd, dccd_stats.blist,
	    dccd_stats.send_error,  dccd_stats.report_retrans);
	if (len >= blen)
		len = blen-1;
	blen -= len;
	plen += len;

	if (!grey_on) {
		len = snprintf(&stats.val.string[plen], blen,
	    L_DWPAT(8)" answers rate-limited "
	    L_DWPAT(4)" anon "
	    L_DWPAT(5)" reports rejected\n",
	    dccd_stats.rl, dccd_stats.anon_rl,  dccd_stats.report_reject);
		if (len >= blen)
			len = blen-1;
		blen -= len;
		plen += len;
	}

	len = snprintf(&stats.val.string[plen], blen,
	    "    %s "
	    L_DWPAT(8)" total flooded in\n"
	    L_DWPAT(8)" accepted "
	    L_DWPAT(6)" stale "
	    L_DWPAT(8)" dup  "
	    L_DWPAT(5)" white    "
	    L_DPAT" delete\n"
	    L_DWPAT(8)" reports added between %s and %s",
	    flod_stats_printf(flod_buf, sizeof(flod_buf),
			      (db_minimum_map || flods_st == FLODS_ST_OFF) ? 0
			      : (flods_st != FLODS_ST_ON) ? 1
			      : 2,
			      oflods.total,
			      oflods.open - oflods_connecting,
			      iflods.open - iflods_connecting),
	    iflod_total,
	    iflod_accepted, iflod_stale, iflod_dup,
	    iflod_wlist, iflod_not_deleted,

	    dccd_stats.adds+db_stats.adds, reset_buf, now_buf);
	if (len >= blen)
		len = blen-1;
	blen -= len;
	plen += len;

	stats.hdr.len = plen + sizeof(stats)-sizeof(stats.val);
	stats.hdr.op = DCC_OP_ADMN;
	send_resp(q, &stats.hdr, 0);
	return 1;
}



void
timestamp_send(const QUEUE *q)
{
	time_t delta;
	DCC_ADMN_RESP msg;
	int blen, plen;

	delta = picky_time(q);

	blen = min(sizeof(msg.val.string), ntohl(q->pkt.ad.val1));
	if (delta < -MAX_CMD_CLOCK_SKEW || delta > MAX_CMD_CLOCK_SKEW) {
		if (delta < -MAX_FLOD_CLOCK_SKEW
		    || delta > MAX_FLOD_CLOCK_SKEW) {
			plen = snprintf(msg.val.string, blen,
					"    clocks differ by about %d seconds"
					"\n    which is more than the"
					" maximum allowed for flooding, %d",
					(int)delta, MAX_FLOD_CLOCK_SKEW);
		} else {
			plen = snprintf(msg.val.string, blen,
					"    clocks differ by about %d seconds"
					"\n    which is more than the"
					" maximum allowed for commands, %d",
					(int)delta, MAX_CMD_CLOCK_SKEW);
		}
	} else {
		plen = snprintf(msg.val.string, blen,
				"    clocks differ by about %d seconds",
				(int)delta);
	}

	msg.hdr.len = plen + sizeof(msg)-sizeof(msg.val);
	msg.hdr.op = DCC_OP_ADMN;
	send_resp(q, &msg.hdr, 0);
}



void
do_nop(QUEUE *q)
{
	/* respond immediately to even anonymous NOPs so that clients
	 * that are confused about passwords and whether they are anonymous
	 * do not retransmit unnecessarily */
	TMSG1(ADMN, "received %s", op_id_ip(q));
	++dccd_stats.nops;

	if (!ck_clnt_srvr_id(q)) {
		++q->rl->d.nops;
		return;
	}

	++q->rl->d.nops;
	send_ok(q);
}



/* deal with an adminstative request */
void
do_admn(QUEUE *q)
{
	u_int32_t val1;
	DCC_ADMN_RESP resp;
	int len, offset;
	u_int32_t adelay_ms;
	struct in6_addr addr6, mask6;
	const struct in6_addr *addr6p, *mask6p;

	val1 = ntohl(q->pkt.ad.val1);
	TMSG3(ADMN, "received val2=%#x val3=%#x in %s",
	      q->pkt.ad.val2, q->pkt.ad.val3, op_id_ip(q));
	++dccd_stats.admin;

	if (!ck_clnt_srvr_id(q))
		return;

	if (q->pkt_len != DCC_ADMN_REQ_MIN_SIZE
	    && (q->pkt_len != (DCC_ADMN_REQ_MIN_SIZE
			       + sizeof(DCC_AOP_CLIENTS_CIDR))
		|| (q->pkt.ad.aop != DCC_AOP_CLIENTS
		    && q->pkt.ad.aop != DCC_AOP_CLIENTS_ID))) {
		send_error(q, "%s size = %d", qop2str(q), q->pkt_len);
		return;
	}

	switch ((DCC_AOPS)q->pkt.ad.aop) {
	case DCC_AOP_STOP:		/* stop gracefully */
		if (!picky_admn(q, 0, 0))
			return;
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (!stopint) {
			stopint = -1;
			next_flods_ck = 0;
		}
		send_ok(q);
		/* fsync() or let the database be wrong	if asked */
		if (val1 != 0)
			stop_mode = val1;
		return;

	case DCC_AOP_DB_UNLOAD:
		if (!picky_admn(q, 0, 0))
			return;
		/* repeat previous answer to repeated question */
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		/* unlike dbclean, dblist starts looking at the data
		 * immediately, so we cannot answer before flushing */
		if (val1 == 0) {
			dcc_trace_msg("database flush started");
			rel_db_states();
			db_minimum_map = 1;
			db_unload(0, 0);
			dcc_trace_msg("database flushed; buffering off");
		} else {
			db_minimum_map = 0;
			dcc_trace_msg("database buffering on");
		}
		send_ok(q);
		return;

	case DCC_AOP_FLOD:		/* control flooding */
		do_flod(q);
		return;

	case DCC_AOP_DB_CLEAN:		/* start switch to new database */
		if (!picky_admn(q, 0, 0))
			return;
		/* repeat previous answer to repeated question */
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (!flods_off || oflods.total != 0) {
			send_error(q, "flooding not stopped before %s",
				   qop2str(q));
			return;
		}
		send_ok(q);		/* asnwer now before we stall */
		dcc_trace_msg("database cleaning begun");
		next_flods_ck = 0;
		/* don't start our own cleaning */
		del_dbclean_next = db_time.tv_sec + DEL_DBCLEAN_SECS;
		dbclean_limit = db_time.tv_sec + dbclean_limit_secs;
		/* Dbclean expects us to remove its separate hold on flooding
		 * so that it will not need to talk to us after telling us
		 * to close the old database.  This because we might stall
		 * on some systems with lame mmap() support including BSD/OS,
		 * for minutes in close().
		 * It might be nice to be able to turn off flooding before
		 * dbclean is run and have it remain off when dbclean
		 * finishes.  However, the need for that that is very rare
		 * and there are mysterious cases where flooding gets
		 * turned off by dbclean and never restored. */
		flods_off = 0;
		/* release and unmap buffers, possibly stalling */
		db_minimum_map = 1;
		rel_db_states();
		db_unload(0, 0);
		return;

	case DCC_AOP_DB_NEW:		/* finish switch to new database */
		if (!picky_admn(q, 0, 0))
			return;
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (!db_minimum_map) {
			send_error(q, "%s received before %s",
				   qop2str(q),
				   dcc_aop2str(0, 0, DCC_AOP_DB_CLEAN, 0));
			return;
		}
		/* send "ok" now because we may stall waiting to reopen */
		send_ok(q);
		db_close(1);
		dccd_stats.adds += db_stats.adds;
		if (!dccd_db_open(DB_OPEN_LOCK_WAIT))
			dcc_logbad(dcc_ex_code,
				   "could not restart database %s: %s",
				   db_nm, dcc_emsg);
		dcc_trace_msg(DCC_VERSION" database %s reopened with %s",
			      db_nm, db_window_size_str);
		flods_off = 0;
		flods_restart("database reopened", 0);
		next_flods_ck = 0;	/* possibly reap dbclean child */
		if (0 >= check_load_ids(2))
			dcc_error_msg("%s", dcc_emsg);
		return;

	case DCC_AOP_STATS:		/* return counters */
		/* we cannot just repeat ourselves for retransmissions,
		 * because the answer is too big to save */
		stats_send(q);
		return;

	case DCC_AOP_STATS_CLEAR:	/* return and then zero counters */
		if (!picky_admn(q, 0, 0))
			return;
		/* we cannot just repeat ourselves for retransmissions,
		 * because the answer is too big to save */
		if (stats_send(q)) {
			clients_clear();
			stats_clear();
		}
		return;

	case DCC_AOP_TRACE_ON:
	case DCC_AOP_TRACE_OFF:
		if (!picky_admn(q, 0, 0))
			return;
		/* it is idempotent, but suppress duplicate trace messages */
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		/* log trace changes even when tracing is off */
		if (!(DCC_TRACE_ADMN_BIT & dccd_tracemask))
			dcc_trace_msg("received %s", op_id_ip(q));
		if ((val1 & ~DCC_TRACE_BITS) != 0 || val1 == 0) {
			send_error(q, "invalid trace bits %#x", val1);
			return;
		}
		if (q->pkt.ad.aop == DCC_AOP_TRACE_OFF) {
			dccd_tracemask &= ~val1;
		} else {
			dccd_tracemask |= val1;
			/* do not suppress the next duplicated flood message */
			if (val1 & DCC_TRACE_FLOD_BIT)
				flod_trace_gen = db_time.tv_sec;
		}
		send_ok(q);
		return;

	case DCC_AOP_CLIENTS:
	case DCC_AOP_CLIENTS_ID:
		if (!picky_admn(q, 1, 1))
			return;
		/* we cannot just repeat ourselves for retransmissions,
		 * because the answer is too big to save */
		offset = (val1 >> 16) + (((u_int)q->pkt.ad.val4) << 16);
		val1 &= 0xffff;
		len = q->pkt.ad.val2;
		if (q->pkt_len == (DCC_ADMN_REQ_MIN_SIZE
				   + sizeof(DCC_AOP_CLIENTS_CIDR))) {
			memcpy(&addr6, &q->pkt.ad.val5[0], sizeof(addr6));
			dcc_bits2mask(&mask6, q->pkt.ad.val5[sizeof(addr6)]);
			addr6p = &addr6;
			mask6p = &mask6;
		} else {
			mask6p = 0;
			addr6p = 0;
		}
		if (q->pkt.ad.aop == DCC_AOP_CLIENTS)
			clients_get(&resp.val, &len, offset,
				    val1, q->pkt.ad.val3, addr6p, mask6p);
		else
			clients_get_id(&resp.val, &len, offset,
				       val1, q->pkt.ad.val3, addr6p, mask6p);
		resp.hdr.len = len + sizeof(resp)-sizeof(resp.val);
		resp.hdr.op = DCC_OP_ADMN;
		send_resp(q, &resp.hdr, 0);
		return;

	case DCC_AOP_ANON_DELAY:
		/* get and set the anonymous client delay
		 *
		 * repeat answer to identical question */
		if (ridc_get(q)) {
			repeat_resp(q);
			return;
		}
		if (anon_off)
			adelay_ms = DCC_ANON_DELAY_FOREVER;
		else
			adelay_ms = anon_delay_us/1000;
		resp.val.anon_delay.delay[0] = adelay_ms>>8;
		resp.val.anon_delay.delay[1] = adelay_ms;
		if (anon_delay_inflate == DCC_ANON_INFLATE_OFF) {
			resp.val.anon_delay.inflate[0] = 0;
			resp.val.anon_delay.inflate[1] = 0;
			resp.val.anon_delay.inflate[2] = 0;
			resp.val.anon_delay.inflate[3] = 0;
		} else {
			resp.val.anon_delay.inflate[0] = anon_delay_inflate>>24;
			resp.val.anon_delay.inflate[1] = anon_delay_inflate>>16;
			resp.val.anon_delay.inflate[2] = anon_delay_inflate>>8;
			resp.val.anon_delay.inflate[3] = anon_delay_inflate;
		}
		adelay_ms = (q->pkt.ad.val2<<8) + q->pkt.ad.val3;
		if (adelay_ms != DCC_NO_ANON_DELAY
		    && picky_admn(q, 0, 0)) {
			if (adelay_ms == DCC_ANON_DELAY_FOREVER) {
				anon_off = 1;
			} else {
				anon_off = 0;
				if (adelay_ms > DCC_ANON_DELAY_MAX/1000)
					adelay_ms = DCC_ANON_DELAY_MAX/1000;
				anon_delay_us = adelay_ms*1000;
				if (val1 == 0)
					val1 = DCC_ANON_INFLATE_OFF;
				anon_delay_inflate = val1;
			}
		}
		resp.hdr.len = (sizeof(resp)-sizeof(resp.val)
				+ sizeof(resp.val.anon_delay));
		resp.hdr.op = DCC_OP_ADMN;
		send_resp(q, &resp.hdr, 0);
		return;

	case DCC_AOP_CLOCK_CHECK:
		timestamp_send(q);
		return;

	case DCC_AOP_OK:
	case DCC_AOP_unused1:
	default:
		break;
	}

	send_error(q, "invalid %s", qop2str(q));
}