ckdb/php check/report -w limits

pool/page_pplns.php

@@ -157,6 +157,9 @@ Block: <input type=text name=blk size=10 value='$blkuse'>
 	if ($ans['ERROR'] != null)
 		return '<font color=red size=+1><br>'.$ans['STATUS'].': '.$ans['ERROR'].'</font>';
 
+	if (!isset($ans['pplns_last']))
+		return '<font color=red size=+1><br>Partial data returned</font>';
+
 	$reward_sat = $ans['block_reward'];
 	$miner_sat = round($reward_sat * 0.991);
 	$ans['miner_sat'] = $miner_sat;
@@ -191,6 +194,13 @@ Block: <input type=text name=blk size=10 value='$blkuse'>
 	$pg = '<br><a href=https://blockchain.info/block-height/';
 	$pg .= $ans['block'].'>Blockchain '.$ans['block']."</a><br>\n";
 
+	if (strlen($ans['marks_status']) > 0)
+	{
+		$pg .= '<br><span class=err>';
+		$msg = $ans['marks_status'];
+		$pg .= str_replace(' ', '&nbsp;', $msg)."</span><br>\n";
+	}
+
 	if (strlen($ans['block_extra']) > 0)
 	{
 		$pg .= '<br><span class=err>';

src/ckdb.c

@@ -154,6 +154,7 @@
 
 static bool socketer_using_data;
 static bool summariser_using_data;
+static bool marker_using_data;
 static bool logger_using_data;
 static bool listener_using_data;
 
@@ -275,6 +276,11 @@ int64_t dbload_workinfoid_finish = MAXID;
 // Only restrict sharesummary, not workinfo
 bool dbload_only_sharesummary = false;
 
+/* If the above restriction - on sharesummaries - is after the last marks
+ *  then this means the sharesummaries can't be summarised into
+ *  markersummaries and pplns payouts may not be correct */
+bool sharesummary_marks_limit = false;
+
 // DB users,workers,auth load is complete
 bool db_auths_complete = false;
 // DB load is complete
@@ -753,6 +759,8 @@ static bool getdata3()
 	}
 	if (!(ok = workinfo_fill(conn)) || everyone_die)
 		goto sukamudai;
+	/* marks must be loaded before sharesummary
+	 * since sharesummary looks at the marks data */
 	if (!(ok = marks_fill(conn)) || everyone_die)
 		goto sukamudai;
 	if (!(ok = workmarkers_fill(conn)) || everyone_die)
@@ -1621,7 +1629,7 @@ static void summarise_blocks()
 	} else {
 		DATA_BLOCKS(prev_blocks, b_prev);
 		wi_start = prev_blocks->workinfoid;
-		wi_item = find_workinfo(wi_start);
+		wi_item = find_workinfo(wi_start, NULL);
 		if (!wi_item) {
 			// This will repeat until fixed ...
 			LOGERR("%s() block %d, but prev %d wid "
@@ -2024,6 +2032,86 @@ static void *summariser(__maybe_unused void *arg)
 	return NULL;
 }
 
+// Number of workinfoids per shift
+#define WID_PER_SHIFT	100
+
+#if 0
+static void make_shift_marks()
+{
+	K_TREE_CTX ctx[1];
+	K_ITEM *m_item, *wi_item;;
+	WORKINFO *workinfo;
+	MARKS *marks;
+	int wid_count;
+	int64_t wid;
+
+	K_RLOCK(marks_free);
+	m_item = last_in_ktree(marks_root, ctx);
+	if (m_item) {
+		DATA_MARKS(marks, m_item);
+		wi_item = find_workinfo(marks->workinfoid, ctx);
+		if (!wi_item) {
+			LOGEMERG("%s() last mark %"PRId64"/%s/%s/%s/%s"
+				 " workinfoid is missing!",
+				 __func__, marks->workinfoid,
+				 marks_marktype(marks->marktype),
+				 marks->status, marks->description,
+				 marks->extra);
+		}
+	}
+	K_RUNLOCK(marks_free);
+}
+#endif
+
+static void *marker(__maybe_unused void *arg)
+{
+	int i;
+
+	pthread_detach(pthread_self());
+
+	rename_proc("db_marker");
+
+	while (!everyone_die && !startup_complete)
+		cksleep_ms(42);
+
+	marker_using_data = true;
+
+	while (!everyone_die) {
+		for (i = 0; i < 5; i++) {
+			if (!everyone_die)
+				sleep(1);
+		}
+#if 0
+		if (everyone_die)
+			break;
+		else
+			make_shift_marks();
+
+		for (i = 0; i < 4; i++) {
+			if (!everyone_die)
+				sleep(1);
+		}
+		if (everyone_die)
+			break;
+		else
+			make_a_workmarker();
+
+		for (i = 0; i < 4; i++) {
+			if (!everyone_die)
+				sleep(1);
+		}
+		if (everyone_die)
+			break;
+		else
+			make_markersummaries();
+#endif
+	}
+
+	marker_using_data = false;
+
+	return NULL;
+}
+
 static void *logger(__maybe_unused void *arg)
 {
 	K_ITEM *lq_item;
@@ -2924,6 +3012,7 @@ static void *listener(void *arg)
 	pthread_t log_pt;
 	pthread_t sock_pt;
 	pthread_t summ_pt;
+	pthread_t mark_pt;
 	K_ITEM *wq_item;
 	time_t now;
 	int wqcount, wqgot;
@@ -2938,6 +3027,8 @@ static void *listener(void *arg)
 
 	create_pthread(&summ_pt, summariser, NULL);
 
+	create_pthread(&mark_pt, marker, NULL);
+
 	rename_proc("db_listener");
 
 	listener_using_data = true;
@@ -3827,7 +3918,8 @@ int main(int argc, char **argv)
 
 	trigger = start = time(NULL);
 	while (socketer_using_data || summariser_using_data ||
-		logger_using_data || listener_using_data) {
+		logger_using_data || listener_using_data ||
+		marker_using_data) {
 		msg = NULL;
 		curr = time(NULL);
 		if (curr - start > 4) {
@@ -3839,12 +3931,13 @@ int main(int argc, char **argv)
 		}
 		if (msg) {
 			trigger = curr;
-			printf("%s %ds due to%s%s%s%s\n",
+			printf("%s %ds due to%s%s%s%s%s\n",
 				msg, (int)(curr - start),
 				socketer_using_data ? " socketer" : EMPTY,
 				summariser_using_data ? " summariser" : EMPTY,
 				logger_using_data ? " logger" : EMPTY,
-				listener_using_data ? " listener" : EMPTY);
+				listener_using_data ? " listener" : EMPTY,
+				marker_using_data ? " marker" : EMPTY);
 			fflush(stdout);
 		}
 		sleep(1);

src/ckdb.h

@@ -52,7 +52,7 @@
 
 #define DB_VLOCK "1"
 #define DB_VERSION "0.9.6"
-#define CKDB_VERSION DB_VERSION"-0.782"
+#define CKDB_VERSION DB_VERSION"-0.800"
 
 #define WHERE_FFL " - from %s %s() line %d"
 #define WHERE_FFL_HERE __FILE__, __func__, __LINE__
@@ -250,6 +250,11 @@ extern int64_t dbload_workinfoid_finish;
 // Only restrict sharesummary, not workinfo
 extern bool dbload_only_sharesummary;
 
+/* If the above restriction - on sharesummaries - is after the last marks
+ *  then this means the sharesummaries can't be summarised into
+ *  markersummaries and pplns payouts may not be correct */
+extern bool sharesummary_marks_limit;
+
 // DB users,workers,auth load is complete
 extern bool db_auths_complete;
 // DB load is complete
@@ -1574,7 +1579,7 @@ extern int32_t _coinbase1height(char *coinbase1, WHERE_FFL_ARGS);
 #define cmp_height(_cb1a, _cb1b) _cmp_height(_cb1a, _cb1b, WHERE_FFL_HERE)
 extern cmp_t _cmp_height(char *coinbase1a, char *coinbase1b, WHERE_FFL_ARGS);
 extern cmp_t cmp_workinfo_height(K_ITEM *a, K_ITEM *b);
-extern K_ITEM *find_workinfo(int64_t workinfoid);
+extern K_ITEM *find_workinfo(int64_t workinfoid, K_TREE_CTX *ctx);
 extern bool workinfo_age(PGconn *conn, int64_t workinfoid, char *poolinstance,
 			 char *by, char *code, char *inet, tv_t *cd,
 			 tv_t *ss_first, tv_t *ss_last, int64_t *ss_count,

src/ckdb_cmd.c

@@ -887,7 +887,7 @@ static char *cmd_blocklist(__maybe_unused PGconn *conn, char *cmd, char *id,
 			snprintf(tmp, sizeof(tmp), "elapsed:%d=%s%c", rows, reply, FLDSEP);
 			APPEND_REALLOC(buf, off, len, tmp);
 
-			w_item = find_workinfo(blocks->workinfoid);
+			w_item = find_workinfo(blocks->workinfoid, NULL);
 			if (w_item) {
 					char wdiffbin[TXT_SML+1];
 					double wdiff;
@@ -3440,7 +3440,8 @@ static char *cmd_pplns(__maybe_unused PGconn *conn, char *cmd, char *id,
 			  __maybe_unused char *code, __maybe_unused char *inet,
 			  __maybe_unused tv_t *notcd, K_TREE *trf_root)
 {
-	char reply[1024], tmp[1024], *buf, *block_extra, *share_status = EMPTY;
+	char reply[1024], tmp[1024], *buf;
+	char *block_extra, *share_status = EMPTY, *marks_status = EMPTY;
 	size_t siz = sizeof(reply);
 	K_ITEM *i_height, *i_difftimes, *i_diffadd, *i_allowaged;
 	K_ITEM b_look, ss_look, *b_item, *w_item, *ss_item;
@@ -3475,6 +3476,9 @@ static char *cmd_pplns(__maybe_unused PGconn *conn, char *cmd, char *id,
 
 	LOGDEBUG("%s(): cmd '%s'", __func__, cmd);
 
+	if (sharesummary_marks_limit)
+		marks_status = "ckdb -w load value means pplns may be incorrect";
+
 	i_height = require_name(trf_root, "height", 1, NULL, reply, siz);
 	if (!i_height)
 		return strdup(reply);
@@ -3543,7 +3547,7 @@ static char *cmd_pplns(__maybe_unused PGconn *conn, char *cmd, char *id,
 			break;
 	}
 	block_workinfoid = blocks->workinfoid;
-	w_item = find_workinfo(block_workinfoid);
+	w_item = find_workinfo(block_workinfoid, NULL);
 	if (!w_item) {
 		snprintf(reply, siz, "ERR.missing workinfo %"PRId64, block_workinfoid);
 		return strdup(reply);
@@ -3705,7 +3709,7 @@ static char *cmd_pplns(__maybe_unused PGconn *conn, char *cmd, char *id,
 		goto shazbot;
 	}
 
-	wb_item = find_workinfo(begin_workinfoid);
+	wb_item = find_workinfo(begin_workinfoid, NULL);
 	if (!wb_item) {
 		snprintf(reply, siz, "ERR.missing begin workinfo record! %"PRId64, block_workinfoid);
 		goto shazbot;
@@ -3737,6 +3741,8 @@ static char *cmd_pplns(__maybe_unused PGconn *conn, char *cmd, char *id,
 	APPEND_REALLOC(buf, off, len, tmp);
 	snprintf(tmp, sizeof(tmp), "share_status=%s%c", share_status, FLDSEP);
 	APPEND_REALLOC(buf, off, len, tmp);
+	snprintf(tmp, sizeof(tmp), "marks_status=%s%c", marks_status, FLDSEP);
+	APPEND_REALLOC(buf, off, len, tmp);
 	snprintf(tmp, sizeof(tmp), "workername=%s%c", blocks->workername, FLDSEP);
 	APPEND_REALLOC(buf, off, len, tmp);
 	snprintf(tmp, sizeof(tmp), "nonce=%s%c", blocks->nonce, FLDSEP);
@@ -3870,8 +3876,10 @@ static char *cmd_pplns(__maybe_unused PGconn *conn, char *cmd, char *id,
 	APPEND_REALLOC(buf, off, len, tmp);
 	snprintf(tmp, sizeof(tmp), "wm_count=%"PRId64"%c", wm_count, FLDSEP);
 	APPEND_REALLOC(buf, off, len, tmp);
-	snprintf(tmp, sizeof(tmp), "ms_count=%"PRId64, ms_count);
+	snprintf(tmp, sizeof(tmp), "ms_count=%"PRId64"%c", ms_count, FLDSEP);
 	APPEND_REALLOC(buf, off, len, tmp);
+	// So web can always verify it received all data
+	APPEND_REALLOC(buf, off, len, "pplns_last=1");
 
 	mu_root = free_ktree(mu_root, NULL);
 	K_WLOCK(mu_store);
@@ -4213,7 +4221,7 @@ static char *cmd_marks(PGconn *conn, char *cmd, char *id,
 			snprintf(reply, siz, "workinfoid not found");
 			return strdup(reply);
 		}
-		w_item = find_workinfo(workinfoid);
+		w_item = find_workinfo(workinfoid, NULL);
 		if (!w_item) {
 			snprintf(reply, siz, "invalid workinfoid %"PRId64,
 				 workinfoid);

src/ckdb_data.c

@@ -1337,12 +1337,15 @@ cmp_t cmp_workinfo_height(K_ITEM *a, K_ITEM *b)
 	return c;
 }
 
-K_ITEM *find_workinfo(int64_t workinfoid)
+K_ITEM *find_workinfo(int64_t workinfoid, K_TREE_CTX *ctx)
 {
 	WORKINFO workinfo;
-	K_TREE_CTX ctx[1];
+	K_TREE_CTX ctx0[1];
 	K_ITEM look, *item;
 
+	if (ctx == NULL)
+		ctx = ctx0;
+
 	workinfo.workinfoid = workinfoid;
 	workinfo.expirydate.tv_sec = default_expiry.tv_sec;
 	workinfo.expirydate.tv_usec = default_expiry.tv_usec;
@@ -1377,7 +1380,7 @@ bool workinfo_age(PGconn *conn, int64_t workinfoid, char *poolinstance,
 	ss_last->tv_sec = ss_last->tv_usec = 0;
 	*ss_count = *s_count = *s_diff = 0;
 
-	wi_item = find_workinfo(workinfoid);
+	wi_item = find_workinfo(workinfoid, NULL);
 	if (!wi_item) {
 		tv_to_buf(cd, cd_buf, sizeof(cd_buf));
 		LOGERR("%s() %"PRId64"/%s/%ld,%ld %.19s no workinfo! Age discarded!",

src/ckdb_dbio.c

@@ -2346,7 +2346,8 @@ bool workinfo_fill(PGconn *conn)
 	char *params[3];
 	int n, i, par = 0;
 	char *field;
-	char *sel;
+	char *sel = NULL;
+	size_t len, off;
 	int fields = 10;
 	bool ok;
 
@@ -2355,14 +2356,31 @@ bool workinfo_fill(PGconn *conn)
 	// TODO: select the data based on sharesummary since old data isn't needed
 	//  however, the ageing rules for workinfo will decide that also
 	//  keep the last block + current? Rules will depend on payout scheme also
-	sel = "select "
+
+	APPEND_REALLOC_INIT(sel, off, len);
+	APPEND_REALLOC(sel, off, len,
+			"select "
 //			"workinfoid,poolinstance,transactiontree,merklehash,prevhash,"
 			"workinfoid,poolinstance,merklehash,prevhash,"
 			"coinbase1,coinbase2,version,bits,ntime,reward"
 			HISTORYDATECONTROL
 			" from workinfo where expirydate=$1 and"
-		" ((workinfoid>=$2 and workinfoid<=$3) or"
+			" ((workinfoid>=$2 and workinfoid<=$3)");
-		"  workinfoid in (select workinfoid from blocks) )";
+
+	// If we aren't loading the full range, ensure the necessary ones are loaded
+	if ((!dbload_only_sharesummary && dbload_workinfoid_start != -1) ||
+	    dbload_workinfoid_finish != MAXID) {
+		APPEND_REALLOC(sel, off, len,
+				// we need all blocks workinfoids
+				" or workinfoid in (select workinfoid from blocks)"
+				// we need all marks workinfoids
+				" or workinfoid in (select workinfoid from marks)"
+				// we need all workmarkers workinfoids (start and end)
+				" or workinfoid in (select workinfoidstart from workmarkers)"
+				" or workinfoid in (select workinfoidend from workmarkers)");
+	}
+	APPEND_REALLOC(sel, off, len, ")");
+
 	par = 0;
 	params[par++] = tv_to_buf((tv_t *)(&default_expiry), NULL, 0);
 	if (dbload_only_sharesummary)
@@ -2550,7 +2568,7 @@ bool shares_add(PGconn *conn, char *workinfoid, char *username, char *workername
 	HISTORYDATEINIT(shares, cd, by, code, inet);
 	HISTORYDATETRANSFER(trf_root, shares);
 
-	wi_item = find_workinfo(shares->workinfoid);
+	wi_item = find_workinfo(shares->workinfoid, NULL);
 	if (!wi_item) {
 		tv_to_buf(cd, cd_buf, sizeof(cd_buf));
 		// TODO: store it for a few workinfoid changes
@@ -2671,7 +2689,7 @@ bool shareerrors_add(PGconn *conn, char *workinfoid, char *username,
 	HISTORYDATEINIT(shareerrors, cd, by, code, inet);
 	HISTORYDATETRANSFER(trf_root, shareerrors);
 
-	wi_item = find_workinfo(shareerrors->workinfoid);
+	wi_item = find_workinfo(shareerrors->workinfoid, NULL);
 	if (!wi_item) {
 		tv_to_buf(cd, cd_buf, sizeof(cd_buf));
 		LOGERR("%s() %"PRId64"/%s/%ld,%ld %.19s no workinfo! Shareerror discarded!",
@@ -3066,9 +3084,11 @@ bool sharesummary_fill(PGconn *conn)
 {
 	ExecStatusType rescode;
 	PGresult *res;
-	K_ITEM *item;
+	K_TREE_CTX ctx[1];
+	K_ITEM *item, *m_item;
 	int n, i, par = 0;
 	SHARESUMMARY *row;
+	MARKS *marks;
 	char *params[2];
 	char *field;
 	char *sel;
@@ -3077,6 +3097,32 @@ bool sharesummary_fill(PGconn *conn)
 
 	LOGDEBUG("%s(): select", __func__);
 
+	/* Load needs to go back to the last marks workinfoid(+1)
+	 * If it is later than that, we can't create markersummaries
+	 *  since some of the required data is missing -
+	 *  thus we also can't make the shift markersummaries */
+	m_item = last_in_ktree(marks_root, ctx);
+	if (!m_item) {
+		if (dbload_workinfoid_start != -1) {
+			sharesummary_marks_limit = true;
+			LOGWARNING("WARNING: dbload -w start used "
+				   "but there are no marks ...");
+		}
+	} else {
+		DATA_MARKS(marks, m_item);
+		if (dbload_workinfoid_start > marks->workinfoid) {
+			sharesummary_marks_limit = true;
+			LOGWARNING("WARNING: dbload -w start %"PRId64
+				   " is after the last mark %"PRId64" ...",
+				   dbload_workinfoid_start,
+				   marks->workinfoid);
+		}
+	}
+	if (sharesummary_marks_limit) {
+		LOGWARNING("WARNING: ... markersummaries cannot be created "
+			   "and pplns calculations may be wrong");
+	}
+
 	// TODO: limit how far back
 	sel = "select "
 		"userid,workername,workinfoid,diffacc,diffsta,diffdup,diffhi,"
@@ -3733,7 +3779,7 @@ flail:
 				break;
 			case BLOCKS_CONFIRM:
 				blk = true;
-				w_item = find_workinfo(row->workinfoid);
+				w_item = find_workinfo(row->workinfoid, NULL);
 				if (w_item) {
 					char wdiffbin[TXT_SML+1];
 					double wdiff;
@@ -5302,10 +5348,10 @@ bool _workmarkers_process(PGconn *conn, bool add, int64_t markerid,
 				 WHERE_FFL_PASS);
 			goto rollback;
 		}
-		w_item = find_workinfo(workinfoidend);
+		w_item = find_workinfo(workinfoidend, NULL);
 		if (!w_item)
 			goto rollback;
-		w_item = find_workinfo(workinfoidstart);
+		w_item = find_workinfo(workinfoidstart, NULL);
 		if (!w_item)
 			goto rollback;
 		K_WLOCK(workmarkers_free);
@@ -5614,7 +5660,7 @@ bool _marks_process(PGconn *conn, bool add, char *poolinstance,
 				 WHERE_FFL_PASS);
 			goto rollback;
 		}
-		w_item = find_workinfo(workinfoid);
+		w_item = find_workinfo(workinfoid, NULL);
 		if (!w_item)
 			goto rollback;
 		K_WLOCK(marks_free);