ckdb - fix esm so that it works and add an esm cmd_query

src/ckdb.c

@@ -2411,10 +2411,11 @@ void sequence_report(bool lock)
 
 static void dealloc_storage()
 {
+	K_ITEM *s_item, *esm_item;
 	SHAREERRORS *shareerrors;
-	K_ITEM *s_item;
 	char *st = NULL;
 	SHARES *shares;
+	ESM *esm;
 	int seq;
 
 	if (free_mode == FREE_MODE_NONE) {
@@ -2541,14 +2542,14 @@ static void dealloc_storage()
 			s_item = STORE_HEAD_NOLOCK(shares_early_store);
 			while (s_item) {
 				DATA_SHARES(shares, s_item);
-				LOGERR("%s(): %"PRId64"/%s/%s/%"PRId32"/%ld,%ld",
-					__func__,
-					shares->workinfoid,
-					st = safe_text_nonull(shares->workername),
-					shares->nonce,
-					shares->errn,
-					shares->createdate.tv_sec,
-					shares->createdate.tv_usec);
+				LOGNOTICE("%s(): %"PRId64"/%s/%s/%"PRId32
+					  "/%ld,%ld", __func__,
+					  shares->workinfoid,
+					  st = safe_text_nonull(shares->workername),
+					  shares->nonce,
+					  shares->errn,
+					  shares->createdate.tv_sec,
+					  shares->createdate.tv_usec);
 				FREENULL(st);
 				s_item = s_item->next;
 			}
@@ -2571,8 +2572,21 @@ static void dealloc_storage()
 	LOGWARNING("%s() etc ...", __func__);
 
 	if (esm_store->count > 0) {
-		LOGWARNING("%s() ***ESM had %d records ...",
-			   __func__, esm_store->count);
+		LOGWARNING("%s() *** ESM had %d record%s ***",
+			   __func__, esm_store->count,
+			   (esm_store->count) == 1 ? EMPTY : "s");
+		esm_item = STORE_HEAD_NOLOCK(esm_store);
+		while (esm_item) {
+			DATA_ESM(esm, esm_item);
+			LOGNOTICE("%s(): %"PRId64" %d/%d/%d err:%d/%d/%d"
+				  " %ld,%ld", __func__,
+				  esm->workinfoid, esm->queued, esm->procured,
+				  esm->discarded, esm->errqueued,
+				  esm->errprocured, esm->errdiscarded,
+				  esm->createdate.tv_sec,
+				  esm->createdate.tv_usec);
+			esm_item = esm_item->next;
+		}
 	}
 	FREE_ALL(esm);
 

src/ckdb.h

@@ -58,7 +58,7 @@
 
 #define DB_VLOCK "1"
 #define DB_VERSION "1.0.7"
-#define CKDB_VERSION DB_VERSION"-2.448"
+#define CKDB_VERSION DB_VERSION"-2.449"
 
 #define WHERE_FFL " - from %s %s() line %d"
 #define WHERE_FFL_HERE __FILE__, __func__, __LINE__

src/ckdb_cmd.c

@@ -7649,6 +7649,52 @@ static char *cmd_query(__maybe_unused PGconn *conn, char *cmd, char *id,
 		K_RUNLOCK(ioqueue_free);
 		rows++;
 
+		ok = true;
+	} else if (strcasecmp(request, "esm") == 0) {
+		K_ITEM *esm_item;
+		ESM *esm = NULL;
+
+		K_RLOCK(esm_free);
+		esm_item = first_in_ktree(esm_root, ctx);
+		while (esm_item) {
+			DATA_ESM(esm, esm_item);
+			snprintf(tmp, sizeof(tmp), "workinfoid:%d=%"PRId64"%c",
+				 rows, esm->workinfoid, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			snprintf(tmp, sizeof(tmp), "queued:%d=%d%c",
+				 rows, esm->queued, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			snprintf(tmp, sizeof(tmp), "procured:%d=%d%c",
+				 rows, esm->procured, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			snprintf(tmp, sizeof(tmp), "discarded:%d=%d%c",
+				 rows, esm->discarded, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			snprintf(tmp, sizeof(tmp), "errqueued:%d=%d%c",
+				 rows, esm->errqueued, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			snprintf(tmp, sizeof(tmp), "errprocured:%d=%d%c",
+				 rows, esm->errprocured, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			snprintf(tmp, sizeof(tmp), "errdiscarded:%d=%d%c",
+				 rows, esm->errdiscarded, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			tv_to_buf(&(esm->createdate), cd_buf, sizeof(cd_buf));
+			snprintf(tmp, sizeof(tmp), CDDB"_str:%d=%s%c",
+				 rows, cd_buf, FLDSEP);
+			APPEND_REALLOC(buf, off, len, tmp);
+			esm_item = next_in_ktree(ctx);
+			rows++;
+		}
+		K_RUNLOCK(esm_free);
+		snprintf(tmp, sizeof(tmp), "flds=%s%c",
+			 "workinfoid,queued,procured,discarded,errqueued,"
+			 "errprocured,errdiscarded,"CDDB"_str", FLDSEP);
+		APPEND_REALLOC(buf, off, len, tmp);
+		snprintf(tmp, sizeof(tmp), "arn=%s%carp=%s%c",
+			 "ESM", FLDSEP, "", FLDSEP);
+		APPEND_REALLOC(buf, off, len, tmp);
+
 		ok = true;
 	} else {
 		free(buf);

src/ckdb_data.c

@@ -2278,6 +2278,8 @@ static bool find_create_esm(int64_t workinfoid, bool error, tv_t *createdate)
 		DATA_ESM(esm, esm_item);
 		esm->workinfoid = workinfoid;
 		copy_tv(&(esm->createdate), createdate);
+		add_to_ktree(esm_root, esm_item);
+		k_add_head(esm_store, esm_item);
 	} else {
 		created = false;
 		DATA_ESM(esm, esm_item);
@@ -2291,13 +2293,22 @@ static bool find_create_esm(int64_t workinfoid, bool error, tv_t *createdate)
 	return created;
 }
 
+/* Early shares are only procured one at a time, with each new share that
+ *  arrives, after the 'late' workinfo arrives
+ * Thus if less shares come in than the number of queued early shares,
+ *  within 60s of the first early share, the DIFF message will appear early
+ *  before the remaining procured messages
+ * This obvioulsy wouldn't happen on a normal running pool
+ * On a small test pool it shouldn't matter since there won't be many extra
+ *  messages */
 void esm_check(tv_t *now)
 {
-	K_ITEM *esm_item = NULL;
+	K_ITEM *esm_item;
 	ESM *esm = NULL;
 	bool had = true;
 
 	while (had) {
+		esm_item = NULL;
 		K_WLOCK(esm_free);
 		if (esm_store->count == 0)
 			had = false;
@@ -2307,6 +2318,7 @@ void esm_check(tv_t *now)
 			while (esm_item) {
 				DATA_ESM(esm, esm_item);
 				if (tvdiff(now, &(esm->createdate)) > ESM_LIMIT) {
+					remove_from_ktree(esm_root, esm_item);
 					k_unlink_item(esm_store, esm_item);
 					break;
 				}