Merge remote-tracking branch 'origin/master' into misc/drop-raft-experiment

lib/harmony/harmonydb/doc.go

@@ -0,0 +1,35 @@
+/*
+# HarmonyDB provides database abstractions over SP-wide Postgres-compatible instance(s).
+
+# Features
+
+	Rolling to secondary database servers on connection failure
+	Convenience features for Go + SQL
+	Prevention of SQL injection vulnerabilities
+	Monitors behavior via Prometheus stats and logging of errors.
+
+# Usage
+
+Processes should use New() to instantiate a *DB and keep it.
+Consumers can use this *DB concurrently.
+Creating and changing tables & views should happen in ./sql/ folder.
+Name the file "today's date" in the format: YYYYMMDD.sql (ex: 20231231.sql for the year's last day)
+
+	a. CREATE TABLE should NOT have a schema:
+		GOOD: CREATE TABLE foo ();
+		BAD:  CREATE TABLE me.foo ();
+	b. Schema is managed for you. It provides isolation for integraton tests & multi-use.
+	c. Git Merges: All run once, so old-after-new is OK when there are no deps.
+	d. NEVER change shipped sql files. Have later files make corrections.
+	e. Anything not ran will be ran, so an older date making it to master is OK.
+
+Write SQL with context, raw strings, and args:
+
+	name := "Alice"
+	var ID int
+	err := QueryRow(ctx, "SELECT id from people where first_name=?", name).Scan(&ID)
+	fmt.Println(ID)
+
+Note: Scan() is column-oriented, while Select() & StructScan() is field name/tag oriented.
+*/
+package harmonydb

lib/harmony/harmonydb/harmonydb.go

@@ -0,0 +1,301 @@
+package harmonydb
+
+import (
+	"context"
+	"embed"
+	"fmt"
+	"math/rand"
+	"net"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/node/config"
+)
+
+type ITestID string
+
+// ItestNewID see ITestWithID doc
+func ITestNewID() ITestID {
+	return ITestID(strconv.Itoa(rand.Intn(99999)))
+}
+
+type DB struct {
+	pgx       *pgxpool.Pool
+	cfg       *pgxpool.Config
+	schema    string
+	hostnames []string
+	BTFPOnce  sync.Once
+	BTFP      atomic.Uintptr
+}
+
+var logger = logging.Logger("harmonydb")
+
+// NewFromConfig is a convenience function.
+// In usage:
+//
+//	db, err := NewFromConfig(config.HarmonyDB)  // in binary init
+func NewFromConfig(cfg config.HarmonyDB) (*DB, error) {
+	return New(
+		cfg.Hosts,
+		cfg.Username,
+		cfg.Password,
+		cfg.Database,
+		cfg.Port,
+		"",
+	)
+}
+
+func NewFromConfigWithITestID(cfg config.HarmonyDB) func(id ITestID) (*DB, error) {
+	return func(id ITestID) (*DB, error) {
+		return New(
+			cfg.Hosts,
+			cfg.Username,
+			cfg.Password,
+			cfg.Database,
+			cfg.Port,
+			id,
+		)
+	}
+}
+
+// New is to be called once per binary to establish the pool.
+// log() is for errors. It returns an upgraded database's connection.
+// This entry point serves both production and integration tests, so it's more DI.
+func New(hosts []string, username, password, database, port string, itestID ITestID) (*DB, error) {
+	itest := string(itestID)
+	connString := ""
+	if len(hosts) > 0 {
+		connString = "host=" + hosts[0] + " "
+	}
+	for k, v := range map[string]string{"user": username, "password": password, "dbname": database, "port": port} {
+		if strings.TrimSpace(v) != "" {
+			connString += k + "=" + v + " "
+		}
+	}
+
+	schema := "lotus"
+	if itest != "" {
+		schema = "itest_" + itest
+	}
+
+	if err := ensureSchemaExists(connString, schema); err != nil {
+		return nil, err
+	}
+	cfg, err := pgxpool.ParseConfig(connString + "search_path=" + schema)
+	if err != nil {
+		return nil, err
+	}
+
+	// enable multiple fallback hosts.
+	for _, h := range hosts[1:] {
+		cfg.ConnConfig.Fallbacks = append(cfg.ConnConfig.Fallbacks, &pgconn.FallbackConfig{Host: h})
+	}
+
+	cfg.ConnConfig.OnNotice = func(conn *pgconn.PgConn, n *pgconn.Notice) {
+		logger.Debug("database notice: " + n.Message + ": " + n.Detail)
+		DBMeasures.Errors.M(1)
+	}
+
+	db := DB{cfg: cfg, schema: schema, hostnames: hosts} // pgx populated in AddStatsAndConnect
+	if err := db.addStatsAndConnect(); err != nil {
+		return nil, err
+	}
+
+	return &db, db.upgrade()
+}
+
+type tracer struct {
+}
+
+type ctxkey string
+
+const SQL_START = ctxkey("sqlStart")
+const SQL_STRING = ctxkey("sqlString")
+
+func (t tracer) TraceQueryStart(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
+	return context.WithValue(context.WithValue(ctx, SQL_START, time.Now()), SQL_STRING, data.SQL)
+}
+func (t tracer) TraceQueryEnd(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryEndData) {
+	DBMeasures.Hits.M(1)
+	ms := time.Since(ctx.Value(SQL_START).(time.Time)).Milliseconds()
+	DBMeasures.TotalWait.M(ms)
+	DBMeasures.Waits.Observe(float64(ms))
+	if data.Err != nil {
+		DBMeasures.Errors.M(1)
+	}
+	logger.Debugw("SQL run",
+		"query", ctx.Value(SQL_STRING).(string),
+		"err", data.Err,
+		"rowCt", data.CommandTag.RowsAffected(),
+		"milliseconds", ms)
+}
+
+func (db *DB) GetRoutableIP() (string, error) {
+	tx, err := db.pgx.Begin(context.Background())
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = tx.Rollback(context.Background()) }()
+	local := tx.Conn().PgConn().Conn().LocalAddr()
+	addr, ok := local.(*net.TCPAddr)
+	if !ok {
+		return "", fmt.Errorf("could not get local addr from %v", addr)
+	}
+	return addr.IP.String(), nil
+}
+
+// addStatsAndConnect connects a prometheus logger. Be sure to run this before using the DB.
+func (db *DB) addStatsAndConnect() error {
+
+	db.cfg.ConnConfig.Tracer = tracer{}
+
+	hostnameToIndex := map[string]float64{}
+	for i, h := range db.hostnames {
+		hostnameToIndex[h] = float64(i)
+	}
+	db.cfg.AfterConnect = func(ctx context.Context, c *pgx.Conn) error {
+		s := db.pgx.Stat()
+		DBMeasures.OpenConnections.M(int64(s.TotalConns()))
+		DBMeasures.WhichHost.Observe(hostnameToIndex[c.Config().Host])
+
+		//FUTURE place for any connection seasoning
+		return nil
+	}
+
+	// Timeout the first connection so we know if the DB is down.
+	ctx, ctxClose := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
+	defer ctxClose()
+	var err error
+	db.pgx, err = pgxpool.NewWithConfig(ctx, db.cfg)
+	if err != nil {
+		logger.Error(fmt.Sprintf("Unable to connect to database: %v\n", err))
+		return err
+	}
+	return nil
+}
+
+// ITestDeleteAll will delete everything created for "this" integration test.
+// This must be called at the end of each integration test.
+func (db *DB) ITestDeleteAll() {
+	if !strings.HasPrefix(db.schema, "itest_") {
+		fmt.Println("Warning: this should never be called on anything but an itest schema.")
+		return
+	}
+	defer db.pgx.Close()
+	_, err := db.pgx.Exec(context.Background(), "DROP SCHEMA "+db.schema+" CASCADE")
+	if err != nil {
+		fmt.Println("warning: unclean itest shutdown: cannot delete schema: " + err.Error())
+		return
+	}
+}
+
+var schemaREString = "^[A-Za-z0-9_]+$"
+var schemaRE = regexp.MustCompile(schemaREString)
+
+func ensureSchemaExists(connString, schema string) error {
+	// FUTURE allow using fallback DBs for start-up.
+	ctx, cncl := context.WithDeadline(context.Background(), time.Now().Add(3*time.Second))
+	p, err := pgx.Connect(ctx, connString)
+	defer cncl()
+	if err != nil {
+		return xerrors.Errorf("unable to connect to db: %s, err: %v", connString, err)
+	}
+	defer func() { _ = p.Close(context.Background()) }()
+
+	if len(schema) < 5 || !schemaRE.MatchString(schema) {
+		return xerrors.New("schema must be of the form " + schemaREString + "\n Got: " + schema)
+	}
+	_, err = p.Exec(context.Background(), "CREATE SCHEMA IF NOT EXISTS "+schema)
+	if err != nil {
+		return xerrors.Errorf("cannot create schema: %w", err)
+	}
+	return nil
+}
+
+//go:embed sql
+var fs embed.FS
+
+func (db *DB) upgrade() error {
+	// Does the version table exist? if not, make it.
+	// NOTE: This cannot change except via the next sql file.
+	_, err := db.Exec(context.Background(), `CREATE TABLE IF NOT EXISTS base (
+		id SERIAL PRIMARY KEY,
+		entry CHAR(12),
+		applied TIMESTAMP DEFAULT current_timestamp
+	)`)
+	if err != nil {
+		logger.Error("Upgrade failed.")
+		return xerrors.Errorf("Cannot create base table %w", err)
+	}
+
+	// __Run scripts in order.__
+
+	landed := map[string]bool{}
+	{
+		var landedEntries []struct{ Entry string }
+		err = db.Select(context.Background(), &landedEntries, "SELECT entry FROM base")
+		if err != nil {
+			logger.Error("Cannot read entries: " + err.Error())
+			return xerrors.Errorf("cannot read entries: %w", err)
+		}
+		for _, l := range landedEntries {
+			landed[l.Entry[:8]] = true
+		}
+	}
+	dir, err := fs.ReadDir("sql")
+	if err != nil {
+		logger.Error("Cannot read fs entries: " + err.Error())
+		return err
+	}
+	sort.Slice(dir, func(i, j int) bool { return dir[i].Name() < dir[j].Name() })
+
+	if len(dir) == 0 {
+		logger.Error("No sql files found.")
+	}
+	for _, e := range dir {
+		name := e.Name()
+		if !strings.HasSuffix(name, ".sql") {
+			logger.Debug("Must have only SQL files here, found: " + name)
+			continue
+		}
+		if landed[name[:8]] {
+			logger.Debug("DB Schema " + name + " already applied.")
+			continue
+		}
+		file, err := fs.ReadFile("sql/" + name)
+		if err != nil {
+			logger.Error("weird embed file read err")
+			return err
+		}
+		for _, s := range strings.Split(string(file), ";") { // Implement the changes.
+			if len(strings.TrimSpace(s)) == 0 {
+				continue
+			}
+			_, err = db.pgx.Exec(context.Background(), s)
+			if err != nil {
+				msg := fmt.Sprintf("Could not upgrade! File %s, Query: %s, Returned: %s", name, s, err.Error())
+				logger.Error(msg)
+				return xerrors.New(msg) // makes devs lives easier by placing message at the end.
+			}
+		}
+
+		// Mark Completed.
+		_, err = db.Exec(context.Background(), "INSERT INTO base (entry) VALUES ($1)", name[:8])
+		if err != nil {
+			logger.Error("Cannot update base: " + err.Error())
+			return xerrors.Errorf("cannot insert into base: %w", err)
+		}
+	}
+	return nil
+}

lib/harmony/harmonydb/metrics.go

@@ -0,0 +1,77 @@
+package harmonydb
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+
+	"github.com/filecoin-project/lotus/metrics"
+)
+
+var (
+	dbTag, _         = tag.NewKey("db_name")
+	pre              = "harmonydb_base_"
+	waitsBuckets     = []float64{0, 10, 20, 30, 50, 80, 130, 210, 340, 550, 890}
+	whichHostBuckets = []float64{0, 1, 2, 3, 4, 5}
+)
+
+// DBMeasures groups all db metrics.
+var DBMeasures = struct {
+	Hits            *stats.Int64Measure
+	TotalWait       *stats.Int64Measure
+	Waits           prometheus.Histogram
+	OpenConnections *stats.Int64Measure
+	Errors          *stats.Int64Measure
+	WhichHost       prometheus.Histogram
+}{
+	Hits:      stats.Int64(pre+"hits", "Total number of uses.", stats.UnitDimensionless),
+	TotalWait: stats.Int64(pre+"total_wait", "Total delay. A numerator over hits to get average wait.", stats.UnitMilliseconds),
+	Waits: prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    pre + "waits",
+		Buckets: waitsBuckets,
+		Help:    "The histogram of waits for query completions.",
+	}),
+	OpenConnections: stats.Int64(pre+"open_connections", "Total connection count.", stats.UnitDimensionless),
+	Errors:          stats.Int64(pre+"errors", "Total error count.", stats.UnitDimensionless),
+	WhichHost: prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    pre + "which_host",
+		Buckets: whichHostBuckets,
+		Help:    "The index of the hostname being used",
+	}),
+}
+
+// CacheViews groups all cache-related default views.
+func init() {
+	metrics.RegisterViews(
+		&view.View{
+			Measure:     DBMeasures.Hits,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+		&view.View{
+			Measure:     DBMeasures.TotalWait,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+		&view.View{
+			Measure:     DBMeasures.OpenConnections,
+			Aggregation: view.LastValue(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+		&view.View{
+			Measure:     DBMeasures.Errors,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+	)
+	err := prometheus.Register(DBMeasures.Waits)
+	if err != nil {
+		panic(err)
+	}
+
+	err = prometheus.Register(DBMeasures.WhichHost)
+	if err != nil {
+		panic(err)
+	}
+}

lib/harmony/harmonydb/sql/20230706.sql

@@ -0,0 +1,7 @@
+CREATE TABLE itest_scratch (
+    id SERIAL PRIMARY KEY,
+	content TEXT,
+    some_int INTEGER,
+    second_int INTEGER,
+	update_time TIMESTAMP DEFAULT current_timestamp
+)

lib/harmony/harmonydb/sql/20230712.sql

@@ -0,0 +1,45 @@
+create table sector_location
+(
+    miner_id         bigint    not null,
+    sector_num       bigint    not null,
+    sector_filetype  int   not null,
+    storage_id       varchar not null,
+    is_primary       bool,
+    read_ts          timestamp(6),
+    read_refs        int,
+    write_ts         timestamp(6),
+    write_lock_owner varchar,
+    constraint sectorlocation_pk
+        primary key (miner_id, sector_num, sector_filetype, storage_id)
+);
+
+alter table sector_location
+    alter column read_refs set not null;
+
+alter table sector_location
+    alter column read_refs set default 0;
+
+create table storage_path
+(
+    "storage_id"  varchar not null
+        constraint "storage_path_pkey"
+            primary key,
+    "urls"       varchar, -- comma separated list of urls
+    "weight"     bigint,
+    "max_storage" bigint,
+    "can_seal"    bool,
+    "can_store"   bool,
+    "groups"     varchar, -- comma separated list of group names
+    "allow_to"    varchar, -- comma separated list of allowed groups
+    "allow_types" varchar,  -- comma separated list of allowed file types
+    "deny_types"  varchar, -- comma separated list of denied file types
+
+    "capacity" bigint,
+    "available" bigint,
+    "fs_available" bigint,
+    "reserved" bigint,
+    "used" bigint,
+    "last_heartbeat" timestamp(6),
+    "heartbeat_err"  varchar
+);
+

lib/harmony/harmonydb/sql/20230719.sql

@@ -0,0 +1,51 @@
+/* For HarmonyTask base implementation. */
+
+CREATE TABLE harmony_machines (
+    id SERIAL PRIMARY KEY NOT NULL,
+    last_contact TIMESTAMP NOT NULL DEFAULT current_timestamp,
+    host_and_port varchar(300) NOT NULL, 
+    cpu INTEGER NOT NULL, 
+    ram BIGINT NOT NULL, 
+    gpu FLOAT NOT NULL
+);
+
+CREATE TABLE harmony_task (
+    id SERIAL PRIMARY KEY NOT NULL,
+    initiated_by INTEGER,     
+    update_time TIMESTAMP NOT NULL DEFAULT current_timestamp,
+    posted_time TIMESTAMP NOT NULL,
+    owner_id INTEGER REFERENCES harmony_machines (id) ON DELETE SET NULL, 
+    added_by INTEGER NOT NULL,
+    previous_task INTEGER,
+    name varchar(16) NOT NULL
+);
+COMMENT ON COLUMN harmony_task.initiated_by IS 'The task ID whose completion occasioned this task.';
+COMMENT ON COLUMN harmony_task.owner_id IS 'The foreign key to harmony_machines.';
+COMMENT ON COLUMN harmony_task.name IS 'The name of the task type.';
+COMMENT ON COLUMN harmony_task.owner_id IS 'may be null if between owners or not yet taken';
+COMMENT ON COLUMN harmony_task.update_time IS 'When it was last modified. not a heartbeat';
+
+CREATE TABLE harmony_task_history (
+    id SERIAL PRIMARY KEY NOT NULL,  
+    task_id INTEGER NOT NULL, 
+    name VARCHAR(16) NOT NULL,
+    posted TIMESTAMP NOT NULL, 
+    work_start TIMESTAMP NOT NULL, 
+    work_end TIMESTAMP NOT NULL, 
+    result BOOLEAN NOT NULL, 
+    err varchar
+);
+COMMENT ON COLUMN harmony_task_history.result IS 'Use to detemine if this was a successful run.';
+
+CREATE TABLE harmony_task_follow (
+    id SERIAL PRIMARY KEY NOT NULL,  
+    owner_id INTEGER NOT NULL REFERENCES harmony_machines (id) ON DELETE CASCADE,
+    to_type VARCHAR(16) NOT NULL,
+    from_type VARCHAR(16) NOT NULL
+);
+
+CREATE TABLE harmony_task_impl (
+    id SERIAL PRIMARY KEY NOT NULL,  
+    owner_id INTEGER NOT NULL REFERENCES harmony_machines (id) ON DELETE CASCADE,
+    name VARCHAR(16) NOT NULL
+);

lib/harmony/harmonydb/sql/20230823.sql

@@ -0,0 +1,48 @@
+create table wdpost_partition_tasks
+(
+    task_id              bigint not null
+        constraint wdpost_partition_tasks_pk
+            primary key,
+    sp_id                bigint not null,
+    proving_period_start bigint not null,
+    deadline_index       bigint not null,
+    partition_index      bigint not null,
+    constraint wdpost_partition_tasks_identity_key
+        unique (sp_id, proving_period_start, deadline_index, partition_index)
+);
+
+comment on column wdpost_partition_tasks.task_id is 'harmonytask task ID';
+comment on column wdpost_partition_tasks.sp_id is 'storage provider ID';
+comment on column wdpost_partition_tasks.proving_period_start is 'proving period start';
+comment on column wdpost_partition_tasks.deadline_index is 'deadline index within the proving period';
+comment on column wdpost_partition_tasks.partition_index is 'partition index within the deadline';
+
+create table wdpost_proofs
+(
+    sp_id                bigint not null,
+    proving_period_start bigint not null,
+    deadline             bigint not null,
+    partition            bigint not null,
+    submit_at_epoch      bigint not null,
+    submit_by_epoch      bigint not null,
+    proof_params         bytea,
+
+    submit_task_id       bigint,
+    message_cid          text,
+
+    constraint wdpost_proofs_identity_key
+        unique (sp_id, proving_period_start, deadline, partition)
+);
+
+create table wdpost_recovery_tasks
+(
+    task_id              bigint not null
+        constraint wdpost_recovery_tasks_pk
+            primary key,
+    sp_id                bigint not null,
+    proving_period_start bigint not null,
+    deadline_index       bigint not null,
+    partition_index      bigint not null,
+    constraint wdpost_recovery_tasks_identity_key
+        unique (sp_id, proving_period_start, deadline_index, partition_index)
+);

lib/harmony/harmonydb/sql/20230919.sql

@@ -0,0 +1,5 @@
+CREATE TABLE harmony_config (
+    id SERIAL PRIMARY KEY NOT NULL,
+    title VARCHAR(300) UNIQUE NOT NULL,
+    config TEXT NOT NULL
+);

lib/harmony/harmonydb/sql/20231103.sql

@@ -0,0 +1,55 @@
+create table message_sends
+(
+    from_key     text   not null,
+    to_addr      text   not null,
+    send_reason  text   not null,
+    send_task_id bigint not null,
+
+    unsigned_data bytea not null,
+    unsigned_cid  text  not null,
+
+    nonce        bigint,
+    signed_data  bytea,
+    signed_json  jsonb,
+    signed_cid   text,
+
+    send_time    timestamp default null,
+    send_success boolean   default null,
+    send_error   text,
+
+    constraint message_sends_pk
+        primary key (send_task_id, from_key)
+);
+
+comment on column message_sends.from_key is 'text f[1/3/4]... address';
+comment on column message_sends.to_addr is 'text f[0/1/2/3/4]... address';
+comment on column message_sends.send_reason is 'optional description of send reason';
+comment on column message_sends.send_task_id is 'harmony task id of the send task';
+
+comment on column message_sends.unsigned_data is 'unsigned message data';
+comment on column message_sends.unsigned_cid is 'unsigned message cid';
+
+comment on column message_sends.nonce is 'assigned message nonce, set while the send task is executing';
+comment on column message_sends.signed_data is 'signed message data, set while the send task is executing';
+comment on column message_sends.signed_cid is 'signed message cid, set while the send task is executing';
+
+comment on column message_sends.send_time is 'time when the send task was executed, set after pushing the message to the network';
+comment on column message_sends.send_success is 'whether this message was broadcasted to the network already, null if not yet attempted, true if successful, false if failed';
+comment on column message_sends.send_error is 'error message if send_success is false';
+
+create unique index message_sends_success_index
+    on message_sends (from_key, nonce)
+    where send_success is not false;
+
+comment on index message_sends_success_index is
+'message_sends_success_index enforces sender/nonce uniqueness, it is a conditional index that only indexes rows where send_success is not false. This allows us to have multiple rows with the same sender/nonce, as long as only one of them was successfully broadcasted (true) to the network or is in the process of being broadcasted (null).';
+
+create table message_send_locks
+(
+    from_key text   not null,
+    task_id  bigint not null,
+    claimed_at timestamp not null,
+
+    constraint message_send_locks_pk
+        primary key (from_key)
+);

lib/harmony/harmonydb/sql/20231110.sql

@@ -0,0 +1,39 @@
+create table mining_tasks
+(
+    task_id bigint not null
+        constraint mining_tasks_pk
+            primary key,
+    sp_id   bigint not null,
+    epoch   bigint not null,
+    base_compute_time timestamp not null,
+
+    won bool not null default false,
+    mined_cid text,
+    mined_header jsonb,
+    mined_at timestamp,
+
+    submitted_at timestamp,
+
+    constraint mining_tasks_sp_epoch
+        unique (sp_id, epoch)
+);
+
+create table mining_base_block
+(
+    id        bigserial not null
+        constraint mining_base_block_pk
+            primary key,
+    task_id   bigint    not null
+        constraint mining_base_block_mining_tasks_task_id_fk
+            references mining_tasks
+            on delete cascade,
+    sp_id     bigint,
+    block_cid text      not null,
+
+    no_win    bool      not null default false,
+
+    constraint mining_base_block_pk2
+        unique (sp_id, task_id, block_cid)
+);
+
+CREATE UNIQUE INDEX mining_base_block_cid_k ON mining_base_block (sp_id, block_cid) WHERE no_win = false;

lib/harmony/harmonydb/sql/20231113.sql

@@ -0,0 +1 @@
+ALTER TABLE harmony_task_history ADD COLUMN completed_by_host_and_port varchar(300) NOT NULL;

lib/harmony/harmonydb/sql/20231120-testing1.sql

@@ -0,0 +1,8 @@
+CREATE TABLE harmony_test (
+    task_id bigint         
+        constraint harmony_test_pk
+            primary key,
+    options text,
+    result text
+);
+ALTER TABLE wdpost_proofs ADD COLUMN test_task_id bigint;

lib/harmony/harmonydb/userfuncs.go

@@ -0,0 +1,205 @@
+package harmonydb
+
+import (
+	"context"
+	"errors"
+	"runtime"
+
+	"github.com/georgysavva/scany/v2/pgxscan"
+	"github.com/jackc/pgerrcode"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/samber/lo"
+)
+
+var errTx = errors.New("Cannot use a non-transaction func in a transaction")
+
+// rawStringOnly is _intentionally_private_ to force only basic strings in SQL queries.
+// In any package, raw strings will satisfy compilation.  Ex:
+//
+//	harmonydb.Exec("INSERT INTO version (number) VALUES (1)")
+//
+// This prevents SQL injection attacks where the input contains query fragments.
+type rawStringOnly string
+
+// Exec executes changes (INSERT, DELETE,  or UPDATE).
+// Note, for CREATE & DROP please keep these permanent and express
+// them in the ./sql/ files (next number).
+func (db *DB) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) (count int, err error) {
+	if db.usedInTransaction() {
+		return 0, errTx
+	}
+	res, err := db.pgx.Exec(ctx, string(sql), arguments...)
+	return int(res.RowsAffected()), err
+}
+
+type Qry interface {
+	Next() bool
+	Err() error
+	Close()
+	Scan(...any) error
+	Values() ([]any, error)
+}
+
+// Query offers Next/Err/Close/Scan/Values/StructScan
+type Query struct {
+	Qry
+}
+
+// Query allows iterating returned values to save memory consumption
+// with the downside of needing to `defer q.Close()`. For a simpler interface,
+// try Select()
+// Next() must be called to advance the row cursor, including the first time:
+// Ex:
+// q, err := db.Query(ctx, "SELECT id, name FROM users")
+// handleError(err)
+// defer q.Close()
+//
+//	for q.Next() {
+//		  var id int
+//	   var name string
+//	   handleError(q.Scan(&id, &name))
+//	   fmt.Println(id, name)
+//	}
+func (db *DB) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error) {
+	if db.usedInTransaction() {
+		return &Query{}, errTx
+	}
+	q, err := db.pgx.Query(ctx, string(sql), arguments...)
+	return &Query{q}, err
+}
+func (q *Query) StructScan(s any) error {
+	return pgxscan.ScanRow(s, q.Qry.(pgx.Rows))
+}
+
+type Row interface {
+	Scan(...any) error
+}
+
+type rowErr struct{}
+
+func (rowErr) Scan(_ ...any) error { return errTx }
+
+// QueryRow gets 1 row using column order matching.
+// This is a timesaver for the special case of wanting the first row returned only.
+// EX:
+//
+//	var name, pet string
+//	var ID = 123
+//	err := db.QueryRow(ctx, "SELECT name, pet FROM users WHERE ID=?", ID).Scan(&name, &pet)
+func (db *DB) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row {
+	if db.usedInTransaction() {
+		return rowErr{}
+	}
+	return db.pgx.QueryRow(ctx, string(sql), arguments...)
+}
+
+/*
+Select multiple rows into a slice using name matching
+Ex:
+
+	type user struct {
+		Name string
+		ID int
+		Number string `db:"tel_no"`
+	}
+
+	var users []user
+	pet := "cat"
+	err := db.Select(ctx, &users, "SELECT name, id, tel_no FROM customers WHERE pet=?", pet)
+*/
+func (db *DB) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error {
+	if db.usedInTransaction() {
+		return errTx
+	}
+	return pgxscan.Select(ctx, db.pgx, sliceOfStructPtr, string(sql), arguments...)
+}
+
+type Tx struct {
+	pgx.Tx
+	ctx context.Context
+}
+
+// usedInTransaction is a helper to prevent nesting transactions
+// & non-transaction calls in transactions. It only checks 20 frames.
+// Fast: This memory should all be in CPU Caches.
+func (db *DB) usedInTransaction() bool {
+	var framePtrs = (&[20]uintptr{})[:]                   // 20 can be stack-local (no alloc)
+	framePtrs = framePtrs[:runtime.Callers(3, framePtrs)] // skip past our caller.
+	return lo.Contains(framePtrs, db.BTFP.Load())         // Unsafe read @ beginTx overlap, but 'return false' is correct there.
+}
+
+// BeginTransaction is how you can access transactions using this library.
+// The entire transaction happens in the function passed in.
+// The return must be true or a rollback will occur.
+// Be sure to test the error for IsErrSerialization() if you want to retry
+//
+//	when there is a DB serialization error.
+//
+//go:noinline
+func (db *DB) BeginTransaction(ctx context.Context, f func(*Tx) (commit bool, err error)) (didCommit bool, retErr error) {
+	db.BTFPOnce.Do(func() {
+		fp := make([]uintptr, 20)
+		runtime.Callers(1, fp)
+		db.BTFP.Store(fp[0])
+	})
+	if db.usedInTransaction() {
+		return false, errTx
+	}
+	tx, err := db.pgx.BeginTx(ctx, pgx.TxOptions{})
+	if err != nil {
+		return false, err
+	}
+	var commit bool
+	defer func() { // Panic clean-up.
+		if !commit {
+			if tmp := tx.Rollback(ctx); tmp != nil {
+				retErr = tmp
+			}
+		}
+	}()
+	commit, err = f(&Tx{tx, ctx})
+	if err != nil {
+		return false, err
+	}
+	if commit {
+		err = tx.Commit(ctx)
+		if err != nil {
+			return false, err
+		}
+		return true, nil
+	}
+	return false, nil
+}
+
+// Exec in a transaction.
+func (t *Tx) Exec(sql rawStringOnly, arguments ...any) (count int, err error) {
+	res, err := t.Tx.Exec(t.ctx, string(sql), arguments...)
+	return int(res.RowsAffected()), err
+}
+
+// Query in a transaction.
+func (t *Tx) Query(sql rawStringOnly, arguments ...any) (*Query, error) {
+	q, err := t.Tx.Query(t.ctx, string(sql), arguments...)
+	return &Query{q}, err
+}
+
+// QueryRow in a transaction.
+func (t *Tx) QueryRow(sql rawStringOnly, arguments ...any) Row {
+	return t.Tx.QueryRow(t.ctx, string(sql), arguments...)
+}
+
+// Select in a transaction.
+func (t *Tx) Select(sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error {
+	return pgxscan.Select(t.ctx, t.Tx, sliceOfStructPtr, string(sql), arguments...)
+}
+
+func IsErrUniqueContraint(err error) bool {
+	var e2 *pgconn.PgError
+	return errors.As(err, &e2) && e2.Code == pgerrcode.UniqueViolation
+}
+
+func IsErrSerialization(err error) bool {
+	var e2 *pgconn.PgError
+	return errors.As(err, &e2) && e2.Code == pgerrcode.SerializationFailure
+}

lib/harmony/harmonytask/doc.go

@@ -0,0 +1,71 @@
+/*
+Package harmonytask implements a pure (no task logic), distributed
+task manager. This clean interface allows a task implementer to completely
+avoid being concerned with task scheduling and management.
+It's based on the idea of tasks as small units of work broken from other
+work by hardware, parallelizabilty, reliability, or any other reason.
+Workers will be Greedy: vaccuuming up their favorite jobs from a list.
+Once 1 task is accepted, harmonydb tries to get other task runner
+machines to accept work (round robin) before trying again to accept.
+*
+Mental Model:
+
+	Things that block tasks:
+		- task not registered for any running server
+		- max was specified and reached
+		- resource exhaustion
+		- CanAccept() interface (per-task implmentation) does not accept it.
+	Ways tasks start:
+		- DB Read every 3 seconds
+		- Task was added (to db) by this process
+	Ways tasks get added:
+	    - Async Listener task (for chain, etc)
+		- Followers: Tasks get added because another task completed
+	When Follower collectors run:
+	    - If both sides are process-local, then this process will pick it up.
+		- If properly registered already, the http endpoint will be tried to start it.
+		- Otherwise, at the listen interval during db scrape it will be found.
+	How duplicate tasks are avoided:
+	    - that's up to the task definition, but probably a unique key
+
+*
+To use:
+1.Implement TaskInterface for a new task.
+2. Have New() receive this & all other ACTIVE implementations.
+*
+*
+As we are not expecting DBAs in this database, it's important to know
+what grows uncontrolled. The only growing harmony_* table is
+harmony_task_history (somewhat quickly). These will need a
+clean-up for after the task data could never be acted upon.
+but the design **requires** extraInfo tables to grow until the task's
+info could not possibly be used by a following task, including slow
+release rollout. This would normally be in the order of months old.
+*
+Other possible enhancements include more collaborative coordination
+to assign a task to machines closer to the data.
+
+__Database_Behavior__
+harmony_task is the list of work that has not been completed.
+
+	AddTaskFunc manages the additions, but is designed to have its
+	transactions failed-out on overlap with a similar task already written.
+	It's up to the TaskInterface implementer to discover this overlap via
+	some other table it uses (since overlap can mean very different things).
+
+harmony_task_history
+
+	This holds transactions that completed or saw too many retries. It also
+	serves as input for subsequent (follower) tasks to kick off. This is not
+	done machine-internally because a follower may not be on the same machine
+	as the previous task.
+
+harmony_task_machines
+
+	Managed by lib/harmony/resources, this is a reference to machines registered
+	via the resources. This registration does not obligate the machine to
+	anything, but serves as a discovery mechanism. Paths are hostnames + ports
+	which are presumed to support http, but this assumption is only used by
+	the task system.
+*/
+package harmonytask

lib/harmony/harmonytask/harmonytask.go

@@ -0,0 +1,307 @@
+package harmonytask
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
+	"github.com/filecoin-project/lotus/lib/harmony/resources"
+)
+
+// Consts (except for unit test)
+var POLL_DURATION = time.Second * 3     // Poll for Work this frequently
+var CLEANUP_FREQUENCY = 5 * time.Minute // Check for dead workers this often * everyone
+var FOLLOW_FREQUENCY = 1 * time.Minute  // Check for work to follow this often
+
+type TaskTypeDetails struct {
+	// Max returns how many tasks this machine can run of this type.
+	// Zero (default) or less means unrestricted.
+	Max int
+
+	// Name is the task name to be added to the task list.
+	Name string
+
+	// Peak costs to Do() the task.
+	Cost resources.Resources
+
+	// Max Failure count before the job is dropped.
+	// 0 = retry forever
+	MaxFailures uint
+
+	// Follow another task's completion via this task's creation.
+	// The function should populate extraInfo from data
+	// available from the previous task's tables, using the given TaskID.
+	// It should also return success if the trigger succeeded.
+	// NOTE: if refatoring tasks, see if your task is
+	// necessary. Ex: Is the sector state correct for your stage to run?
+	Follows map[string]func(TaskID, AddTaskFunc) (bool, error)
+}
+
+// TaskInterface must be implemented in order to have a task used by harmonytask.
+type TaskInterface interface {
+	// Do the task assigned. Call stillOwned before making single-writer-only
+	// changes to ensure the work has not been stolen.
+	// This is the ONLY function that should attempt to do the work, and must
+	// ONLY be called by harmonytask.
+	// Indicate if the task no-longer needs scheduling with done=true including
+	// cases where it's past the deadline.
+	Do(taskID TaskID, stillOwned func() bool) (done bool, err error)
+
+	// CanAccept should return if the task can run on this machine. It should
+	// return null if the task type is not allowed on this machine.
+	// It should select the task it most wants to accomplish.
+	// It is also responsible for determining & reserving disk space (including scratch).
+	CanAccept([]TaskID, *TaskEngine) (*TaskID, error)
+
+	// TypeDetails() returns static details about how this task behaves and
+	// how this machine will run it. Read once at the beginning.
+	TypeDetails() TaskTypeDetails
+
+	// This listener will consume all external sources continuously for work.
+	// Do() may also be called from a backlog of work. This must not
+	// start doing the work (it still must be scheduled).
+	// Note: Task de-duplication should happen in ExtraInfoFunc by
+	//  returning false, typically by determining from the tx that the work
+	//  exists already. The easy way is to have a unique joint index
+	//  across all fields that will be common.
+	// Adder should typically only add its own task type, but multiple
+	//   is possible for when 1 trigger starts 2 things.
+	// Usage Example:
+	// func (b *BazType)Adder(addTask AddTaskFunc) {
+	//	  for {
+	//      bazMaker := <- bazChannel
+	//	    addTask("baz", func(t harmonytask.TaskID, txn db.Transaction) (bool, error) {
+	//	       _, err := txn.Exec(`INSERT INTO bazInfoTable (taskID, qix, mot)
+	//			  VALUES ($1,$2,$3)`, id, bazMaker.qix, bazMaker.mot)
+	//         if err != nil {
+	//				scream(err)
+	//	 		 	return false
+	//		   }
+	// 		   return true
+	//		})
+	//	  }
+	// }
+	Adder(AddTaskFunc)
+}
+
+// AddTaskFunc is responsible for adding a task's details "extra info" to the DB.
+// It should return true if the task should be added, false if it was already there.
+// This is typically accomplished with a "unique" index on your detals table that
+// would cause the insert to fail.
+// The error indicates that instead of a conflict (which we should ignore) that we
+// actually have a serious problem that needs to be logged with context.
+type AddTaskFunc func(extraInfo func(TaskID, *harmonydb.Tx) (shouldCommit bool, seriousError error))
+
+type TaskEngine struct {
+	ctx            context.Context
+	handlers       []*taskTypeHandler
+	db             *harmonydb.DB
+	reg            *resources.Reg
+	grace          context.CancelFunc
+	taskMap        map[string]*taskTypeHandler
+	ownerID        int
+	follows        map[string][]followStruct
+	lastFollowTime time.Time
+	lastCleanup    atomic.Value
+	hostAndPort    string
+}
+type followStruct struct {
+	f    func(TaskID, AddTaskFunc) (bool, error)
+	h    *taskTypeHandler
+	name string
+}
+
+type TaskID int
+
+// New creates all the task definitions. Note that TaskEngine
+// knows nothing about the tasks themselves and serves to be a
+// generic container for common work
+func New(
+	db *harmonydb.DB,
+	impls []TaskInterface,
+	hostnameAndPort string) (*TaskEngine, error) {
+
+	reg, err := resources.Register(db, hostnameAndPort)
+	if err != nil {
+		return nil, fmt.Errorf("cannot get resources: %w", err)
+	}
+	ctx, grace := context.WithCancel(context.Background())
+	e := &TaskEngine{
+		ctx:         ctx,
+		grace:       grace,
+		db:          db,
+		reg:         reg,
+		ownerID:     reg.Resources.MachineID, // The current number representing "hostAndPort"
+		taskMap:     make(map[string]*taskTypeHandler, len(impls)),
+		follows:     make(map[string][]followStruct),
+		hostAndPort: hostnameAndPort,
+	}
+	e.lastCleanup.Store(time.Now())
+	for _, c := range impls {
+		h := taskTypeHandler{
+			TaskInterface:   c,
+			TaskTypeDetails: c.TypeDetails(),
+			TaskEngine:      e,
+		}
+
+		if len(h.Name) > 16 {
+			return nil, fmt.Errorf("task name too long: %s, max 16 characters", h.Name)
+		}
+
+		e.handlers = append(e.handlers, &h)
+		e.taskMap[h.TaskTypeDetails.Name] = &h
+	}
+
+	// resurrect old work
+	{
+		var taskRet []struct {
+			ID   int
+			Name string
+		}
+
+		err := db.Select(e.ctx, &taskRet, `SELECT id, name from harmony_task WHERE owner_id=$1`, e.ownerID)
+		if err != nil {
+			return nil, err
+		}
+		for _, w := range taskRet {
+			// edge-case: if old assignments are not available tasks, unlock them.
+			h := e.taskMap[w.Name]
+			if h == nil {
+				_, err := db.Exec(e.ctx, `UPDATE harmony_task SET owner=NULL WHERE id=$1`, w.ID)
+				if err != nil {
+					log.Errorw("Cannot remove self from owner field", "error", err)
+					continue // not really fatal, but not great
+				}
+			}
+			if !h.considerWork(workSourceRecover, []TaskID{TaskID(w.ID)}) {
+				log.Error("Strange: Unable to accept previously owned task: ", w.ID, w.Name)
+			}
+		}
+	}
+	for _, h := range e.handlers {
+		go h.Adder(h.AddTask)
+	}
+	go e.poller()
+
+	return e, nil
+}
+
+// GracefullyTerminate hangs until all present tasks have completed.
+// Call this to cleanly exit the process. As some processes are long-running,
+// passing a deadline will ignore those still running (to be picked-up later).
+func (e *TaskEngine) GracefullyTerminate(deadline time.Duration) {
+	e.grace()
+	e.reg.Shutdown()
+	deadlineChan := time.NewTimer(deadline).C
+top:
+	for _, h := range e.handlers {
+		if h.Count.Load() > 0 {
+			select {
+			case <-deadlineChan:
+				return
+			default:
+				time.Sleep(time.Millisecond)
+				goto top
+			}
+		}
+	}
+}
+
+func (e *TaskEngine) poller() {
+	for {
+		select {
+		case <-time.NewTicker(POLL_DURATION).C: // Find work periodically
+		case <-e.ctx.Done(): ///////////////////// Graceful exit
+			return
+		}
+		e.pollerTryAllWork()
+		if time.Since(e.lastFollowTime) > FOLLOW_FREQUENCY {
+			e.followWorkInDB()
+		}
+	}
+}
+
+// followWorkInDB implements "Follows"
+func (e *TaskEngine) followWorkInDB() {
+	// Step 1: What are we following?
+	var lastFollowTime time.Time
+	lastFollowTime, e.lastFollowTime = e.lastFollowTime, time.Now()
+
+	for fromName, srcs := range e.follows {
+		var cList []int // Which work is done (that we follow) since we last checked?
+		err := e.db.Select(e.ctx, &cList, `SELECT h.task_id FROM harmony_task_history 
+   		WHERE h.work_end>$1 AND h.name=$2`, lastFollowTime, fromName)
+		if err != nil {
+			log.Error("Could not query DB: ", err)
+			return
+		}
+		for _, src := range srcs {
+			for _, workAlreadyDone := range cList { // Were any tasks made to follow these tasks?
+				var ct int
+				err := e.db.QueryRow(e.ctx, `SELECT COUNT(*) FROM harmony_task 
+					WHERE name=$1 AND previous_task=$2`, src.h.Name, workAlreadyDone).Scan(&ct)
+				if err != nil {
+					log.Error("Could not query harmony_task: ", err)
+					return // not recoverable here
+				}
+				if ct > 0 {
+					continue
+				}
+				// we need to create this task
+				b, err := src.h.Follows[fromName](TaskID(workAlreadyDone), src.h.AddTask)
+				if err != nil {
+					log.Errorw("Could not follow: ", "error", err)
+					continue
+				}
+				if !b {
+					// But someone may have beaten us to it.
+					log.Debugf("Unable to add task %s following Task(%d, %s)", src.h.Name, workAlreadyDone, fromName)
+				}
+			}
+		}
+	}
+}
+
+// pollerTryAllWork starts the next 1 task
+func (e *TaskEngine) pollerTryAllWork() {
+	if time.Since(e.lastCleanup.Load().(time.Time)) > CLEANUP_FREQUENCY {
+		e.lastCleanup.Store(time.Now())
+		resources.CleanupMachines(e.ctx, e.db)
+	}
+	for _, v := range e.handlers {
+		if v.AssertMachineHasCapacity() != nil {
+			continue
+		}
+		var unownedTasks []TaskID
+		err := e.db.Select(e.ctx, &unownedTasks, `SELECT id 
+			FROM harmony_task
+			WHERE owner_id IS NULL AND name=$1
+			ORDER BY update_time`, v.Name)
+		if err != nil {
+			log.Error("Unable to read work ", err)
+			continue
+		}
+		if len(unownedTasks) > 0 {
+			accepted := v.considerWork(workSourcePoller, unownedTasks)
+			if accepted {
+				return // accept new work slowly and in priority order
+			}
+			log.Warn("Work not accepted for " + strconv.Itoa(len(unownedTasks)) + " " + v.Name + " task(s)")
+		}
+	}
+}
+
+// ResourcesAvailable determines what resources are still unassigned.
+func (e *TaskEngine) ResourcesAvailable() resources.Resources {
+	tmp := e.reg.Resources
+	for _, t := range e.handlers {
+		ct := t.Count.Load()
+		tmp.Cpu -= int(ct) * t.Cost.Cpu
+		tmp.Gpu -= float64(ct) * t.Cost.Gpu
+		tmp.Ram -= uint64(ct) * t.Cost.Ram
+	}
+	return tmp
+}

lib/harmony/harmonytask/task_type_handler.go

@@ -0,0 +1,254 @@
+package harmonytask
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	logging "github.com/ipfs/go-log/v2"
+
+	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
+)
+
+var log = logging.Logger("harmonytask")
+
+type taskTypeHandler struct {
+	TaskInterface
+	TaskTypeDetails
+	TaskEngine *TaskEngine
+	Count      atomic.Int32
+}
+
+func (h *taskTypeHandler) AddTask(extra func(TaskID, *harmonydb.Tx) (bool, error)) {
+	var tID TaskID
+	retryWait := time.Millisecond * 100
+retryAddTask:
+	_, err := h.TaskEngine.db.BeginTransaction(h.TaskEngine.ctx, func(tx *harmonydb.Tx) (bool, error) {
+		// create taskID (from DB)
+		_, err := tx.Exec(`INSERT INTO harmony_task (name, added_by, posted_time) 
+			VALUES ($1, $2, CURRENT_TIMESTAMP) `, h.Name, h.TaskEngine.ownerID)
+		if err != nil {
+			return false, fmt.Errorf("could not insert into harmonyTask: %w", err)
+		}
+		err = tx.QueryRow("SELECT id FROM harmony_task ORDER BY update_time DESC LIMIT 1").Scan(&tID)
+		if err != nil {
+			return false, fmt.Errorf("Could not select ID: %v", err)
+		}
+		return extra(tID, tx)
+	})
+
+	if err != nil {
+		if harmonydb.IsErrUniqueContraint(err) {
+			log.Debugf("addtask(%s) saw unique constraint, so it's added already.", h.Name)
+			return
+		}
+		if harmonydb.IsErrSerialization(err) {
+			time.Sleep(retryWait)
+			retryWait *= 2
+			goto retryAddTask
+		}
+		log.Error("Could not add task. AddTasFunc failed: %v", err)
+		return
+	}
+}
+
+const (
+	workSourcePoller  = "poller"
+	workSourceRecover = "recovered"
+)
+
+// considerWork is called to attempt to start work on a task-id of this task type.
+// It presumes single-threaded calling, so there should not be a multi-threaded re-entry.
+// The only caller should be the one work poller thread. This does spin off other threads,
+// but those should not considerWork. Work completing may lower the resource numbers
+// unexpectedly, but that will not invalidate work being already able to fit.
+func (h *taskTypeHandler) considerWork(from string, ids []TaskID) (workAccepted bool) {
+top:
+	if len(ids) == 0 {
+		return true // stop looking for takers
+	}
+
+	// 1. Can we do any more of this task type?
+	// NOTE: 0 is the default value, so this way people don't need to worry about
+	// this setting unless they want to limit the number of tasks of this type.
+	if h.Max > 0 && int(h.Count.Load()) >= h.Max {
+		log.Debugw("did not accept task", "name", h.Name, "reason", "at max already")
+		return false
+	}
+
+	// 2. Can we do any more work? From here onward, we presume the resource
+	// story will not change, so single-threaded calling is best.
+	err := h.AssertMachineHasCapacity()
+	if err != nil {
+		log.Debugw("did not accept task", "name", h.Name, "reason", "at capacity already: "+err.Error())
+		return false
+	}
+
+	// 3. What does the impl say?
+	tID, err := h.CanAccept(ids, h.TaskEngine)
+	if err != nil {
+		log.Error(err)
+		return false
+	}
+	if tID == nil {
+		log.Infow("did not accept task", "task_id", ids[0], "reason", "CanAccept() refused", "name", h.Name)
+		return false
+	}
+
+	// if recovering we don't need to try to claim anything because those tasks are already claimed by us
+	if from != workSourceRecover {
+		// 4. Can we claim the work for our hostname?
+		ct, err := h.TaskEngine.db.Exec(h.TaskEngine.ctx, "UPDATE harmony_task SET owner_id=$1 WHERE id=$2 AND owner_id IS NULL", h.TaskEngine.ownerID, *tID)
+		if err != nil {
+			log.Error(err)
+			return false
+		}
+		if ct == 0 {
+			log.Infow("did not accept task", "task_id", strconv.Itoa(int(*tID)), "reason", "already Taken", "name", h.Name)
+			var tryAgain = make([]TaskID, 0, len(ids)-1)
+			for _, id := range ids {
+				if id != *tID {
+					tryAgain = append(tryAgain, id)
+				}
+			}
+			ids = tryAgain
+			goto top
+		}
+	}
+
+	h.Count.Add(1)
+	go func() {
+		log.Infow("Beginning work on Task", "id", *tID, "from", from, "name", h.Name)
+
+		var done bool
+		var doErr error
+		workStart := time.Now()
+
+		defer func() {
+			if r := recover(); r != nil {
+				stackSlice := make([]byte, 4092)
+				sz := runtime.Stack(stackSlice, false)
+				log.Error("Recovered from a serious error "+
+					"while processing "+h.Name+" task "+strconv.Itoa(int(*tID))+": ", r,
+					" Stack: ", string(stackSlice[:sz]))
+			}
+			h.Count.Add(-1)
+
+			h.recordCompletion(*tID, workStart, done, doErr)
+			if done {
+				for _, fs := range h.TaskEngine.follows[h.Name] { // Do we know of any follows for this task type?
+					if _, err := fs.f(*tID, fs.h.AddTask); err != nil {
+						log.Error("Could not follow", "error", err, "from", h.Name, "to", fs.name)
+					}
+				}
+			}
+		}()
+
+		done, doErr = h.Do(*tID, func() bool {
+			var owner int
+			// Background here because we don't want GracefulRestart to block this save.
+			err := h.TaskEngine.db.QueryRow(context.Background(),
+				`SELECT owner_id FROM harmony_task WHERE id=$1`, *tID).Scan(&owner)
+			if err != nil {
+				log.Error("Cannot determine ownership: ", err)
+				return false
+			}
+			return owner == h.TaskEngine.ownerID
+		})
+		if doErr != nil {
+			log.Errorw("Do() returned error", "type", h.Name, "id", strconv.Itoa(int(*tID)), "error", doErr)
+		}
+	}()
+	return true
+}
+
+func (h *taskTypeHandler) recordCompletion(tID TaskID, workStart time.Time, done bool, doErr error) {
+	workEnd := time.Now()
+	retryWait := time.Millisecond * 100
+retryRecordCompletion:
+	cm, err := h.TaskEngine.db.BeginTransaction(h.TaskEngine.ctx, func(tx *harmonydb.Tx) (bool, error) {
+		var postedTime time.Time
+		err := tx.QueryRow(`SELECT posted_time FROM harmony_task WHERE id=$1`, tID).Scan(&postedTime)
+		if err != nil {
+			return false, fmt.Errorf("could not log completion: %w ", err)
+		}
+		result := "unspecified error"
+		if done {
+			_, err = tx.Exec("DELETE FROM harmony_task WHERE id=$1", tID)
+			if err != nil {
+
+				return false, fmt.Errorf("could not log completion: %w", err)
+			}
+			result = ""
+			if doErr != nil {
+				result = "non-failing error: " + doErr.Error()
+			}
+		} else {
+			if doErr != nil {
+				result = "error: " + doErr.Error()
+			}
+			var deleteTask bool
+			if h.MaxFailures > 0 {
+				ct := uint(0)
+				err = tx.QueryRow(`SELECT count(*) FROM harmony_task_history 
+				WHERE task_id=$1 AND result=FALSE`, tID).Scan(&ct)
+				if err != nil {
+					return false, fmt.Errorf("could not read task history: %w", err)
+				}
+				if ct >= h.MaxFailures {
+					deleteTask = true
+				}
+			}
+			if deleteTask {
+				_, err = tx.Exec("DELETE FROM harmony_task WHERE id=$1", tID)
+				if err != nil {
+					return false, fmt.Errorf("could not delete failed job: %w", err)
+				}
+				// Note: Extra Info is left laying around for later review & clean-up
+			} else {
+				_, err := tx.Exec(`UPDATE harmony_task SET owner_id=NULL WHERE id=$1`, tID)
+				if err != nil {
+					return false, fmt.Errorf("could not disown failed task: %v %v", tID, err)
+				}
+			}
+		}
+		_, err = tx.Exec(`INSERT INTO harmony_task_history 
+									 (task_id,   name, posted,    work_start, work_end, result, completed_by_host_and_port,      err)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, tID, h.Name, postedTime, workStart, workEnd, done, h.TaskEngine.hostAndPort, result)
+		if err != nil {
+			return false, fmt.Errorf("could not write history: %w", err)
+		}
+		return true, nil
+	})
+	if err != nil {
+		if harmonydb.IsErrSerialization(err) {
+			time.Sleep(retryWait)
+			retryWait *= 2
+			goto retryRecordCompletion
+		}
+		log.Error("Could not record transaction: ", err)
+		return
+	}
+	if !cm {
+		log.Error("Committing the task records failed")
+	}
+}
+
+func (h *taskTypeHandler) AssertMachineHasCapacity() error {
+	r := h.TaskEngine.ResourcesAvailable()
+
+	if r.Cpu-h.Cost.Cpu < 0 {
+		return errors.New("Did not accept " + h.Name + " task: out of cpu")
+	}
+	if h.Cost.Ram > r.Ram {
+		return errors.New("Did not accept " + h.Name + " task: out of RAM")
+	}
+	if r.Gpu-h.Cost.Gpu < 0 {
+		return errors.New("Did not accept " + h.Name + " task: out of available GPU")
+	}
+	return nil
+}

lib/harmony/resources/getGPU.go

@@ -0,0 +1,22 @@
+//go:build !darwin
+// +build !darwin
+
+package resources
+
+import (
+	"strings"
+
+	ffi "github.com/filecoin-project/filecoin-ffi"
+)
+
+func getGPUDevices() float64 { // GPU boolean
+	gpus, err := ffi.GetGPUDevices()
+	if err != nil {
+		logger.Errorf("getting gpu devices failed: %+v", err)
+	}
+	all := strings.ToLower(strings.Join(gpus, ","))
+	if len(gpus) > 1 || strings.Contains(all, "ati") || strings.Contains(all, "nvidia") {
+		return float64(len(gpus))
+	}
+	return 0
+}

lib/harmony/resources/getGPU_darwin.go

@@ -0,0 +1,8 @@
+//go:build darwin
+// +build darwin
+
+package resources
+
+func getGPUDevices() float64 {
+	return 10000.0 // unserious value intended for non-production use.
+}

lib/harmony/resources/memsys.go

@@ -0,0 +1,22 @@
+//go:build darwin || freebsd || openbsd || dragonfly || netbsd
+// +build darwin freebsd openbsd dragonfly netbsd
+
+package resources
+
+import (
+	"encoding/binary"
+	"syscall"
+)
+
+func sysctlUint64(name string) (uint64, error) {
+	s, err := syscall.Sysctl(name)
+	if err != nil {
+		return 0, err
+	}
+	// hack because the string conversion above drops a \0
+	b := []byte(s)
+	if len(b) < 8 {
+		b = append(b, 0)
+	}
+	return binary.LittleEndian.Uint64(b), nil
+}

lib/harmony/resources/miniopencl/cl.h

@@ -0,0 +1,17 @@
+  
+#ifndef CL_H
+#define CL_H
+
+#define CL_USE_DEPRECATED_OPENCL_1_1_APIS
+#define CL_USE_DEPRECATED_OPENCL_1_2_APIS
+#define CL_USE_DEPRECATED_OPENCL_2_0_APIS
+
+#define CL_TARGET_OPENCL_VERSION 300
+
+#ifdef __APPLE__
+#include "OpenCL/opencl.h"
+#else
+#include "CL/opencl.h"
+#endif
+
+#endif /* CL_H */

lib/harmony/resources/miniopencl/mini_opencl.go

@@ -0,0 +1,93 @@
+// Package cl was borrowed from the go-opencl library which is more complex and
+// doesn't compile well for our needs.
+package cl
+
+// #include "cl.h"
+import "C"
+import (
+	"fmt"
+	"unsafe"
+)
+
+const maxPlatforms = 32
+
+type Platform struct {
+	id C.cl_platform_id
+}
+
+// Obtain the list of platforms available.
+func GetPlatforms() ([]*Platform, error) {
+	var platformIds [maxPlatforms]C.cl_platform_id
+	var nPlatforms C.cl_uint
+	err := C.clGetPlatformIDs(C.cl_uint(maxPlatforms), &platformIds[0], &nPlatforms)
+	if err == -1001 { // No platforms found
+		return nil, nil
+	}
+	if err != C.CL_SUCCESS {
+		return nil, toError(err)
+	}
+	platforms := make([]*Platform, nPlatforms)
+	for i := 0; i < int(nPlatforms); i++ {
+		platforms[i] = &Platform{id: platformIds[i]}
+	}
+	return platforms, nil
+}
+
+const maxDeviceCount = 64
+
+type DeviceType uint
+
+const (
+	DeviceTypeAll DeviceType = C.CL_DEVICE_TYPE_ALL
+)
+
+type Device struct {
+	id C.cl_device_id
+}
+
+func (p *Platform) GetAllDevices() ([]*Device, error) {
+	var deviceIds [maxDeviceCount]C.cl_device_id
+	var numDevices C.cl_uint
+	var platformId C.cl_platform_id
+	if p != nil {
+		platformId = p.id
+	}
+	if err := C.clGetDeviceIDs(platformId, C.cl_device_type(DeviceTypeAll), C.cl_uint(maxDeviceCount), &deviceIds[0], &numDevices); err != C.CL_SUCCESS {
+		return nil, toError(err)
+	}
+	if numDevices > maxDeviceCount {
+		numDevices = maxDeviceCount
+	}
+	devices := make([]*Device, numDevices)
+	for i := 0; i < int(numDevices); i++ {
+		devices[i] = &Device{id: deviceIds[i]}
+	}
+	return devices, nil
+}
+
+func toError(code C.cl_int) error {
+	return ErrOther(code)
+}
+
+type ErrOther int
+
+func (e ErrOther) Error() string {
+	return fmt.Sprintf("OpenCL: error %d", int(e))
+}
+
+// Size of global device memory in bytes.
+func (d *Device) GlobalMemSize() int64 {
+	val, _ := d.getInfoUlong(C.CL_DEVICE_GLOBAL_MEM_SIZE, true)
+	return val
+}
+
+func (d *Device) getInfoUlong(param C.cl_device_info, panicOnError bool) (int64, error) {
+	var val C.cl_ulong
+	if err := C.clGetDeviceInfo(d.id, param, C.size_t(unsafe.Sizeof(val)), unsafe.Pointer(&val), nil); err != C.CL_SUCCESS {
+		if panicOnError {
+			panic("Should never fail")
+		}
+		return 0, toError(err)
+	}
+	return int64(val), nil
+}

lib/harmony/resources/resources.go

@@ -0,0 +1,142 @@
+package resources
+
+import (
+	"bytes"
+	"context"
+	"os/exec"
+	"regexp"
+	"runtime"
+	"sync/atomic"
+	"time"
+
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/pbnjay/memory"
+	"golang.org/x/sys/unix"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
+)
+
+var LOOKS_DEAD_TIMEOUT = 10 * time.Minute // Time w/o minute heartbeats
+
+type Resources struct {
+	Cpu       int
+	Gpu       float64
+	Ram       uint64
+	MachineID int
+}
+type Reg struct {
+	Resources
+	shutdown atomic.Bool
+}
+
+var logger = logging.Logger("harmonytask")
+
+var lotusRE = regexp.MustCompile("lotus-worker|lotus-harmony|yugabyted|yb-master|yb-tserver")
+
+func Register(db *harmonydb.DB, hostnameAndPort string) (*Reg, error) {
+	var reg Reg
+	var err error
+	reg.Resources, err = getResources()
+	if err != nil {
+		return nil, err
+	}
+	ctx := context.Background()
+	{ // Learn our owner_id while updating harmony_machines
+		var ownerID *int
+
+		// Upsert query with last_contact update, fetch the machine ID
+		// (note this isn't a simple insert .. on conflict because host_and_port isn't unique)
+		err := db.QueryRow(ctx, `
+			WITH upsert AS (
+				UPDATE harmony_machines
+				SET cpu = $2, ram = $3, gpu = $4, last_contact = CURRENT_TIMESTAMP
+				WHERE host_and_port = $1
+				RETURNING id
+			),
+			inserted AS (
+				INSERT INTO harmony_machines (host_and_port, cpu, ram, gpu, last_contact)
+				SELECT $1, $2, $3, $4, CURRENT_TIMESTAMP
+				WHERE NOT EXISTS (SELECT id FROM upsert)
+				RETURNING id
+			)
+			SELECT id FROM upsert
+			UNION ALL
+			SELECT id FROM inserted;
+		`, hostnameAndPort, reg.Cpu, reg.Ram, reg.Gpu).Scan(&ownerID)
+		if err != nil {
+			return nil, xerrors.Errorf("inserting machine entry: %w", err)
+		}
+		if ownerID == nil {
+			return nil, xerrors.Errorf("no owner id")
+		}
+
+		reg.MachineID = *ownerID
+
+		cleaned := CleanupMachines(context.Background(), db)
+		logger.Infow("Cleaned up machines", "count", cleaned)
+	}
+	go func() {
+		for {
+			time.Sleep(time.Minute)
+			if reg.shutdown.Load() {
+				return
+			}
+			_, err := db.Exec(ctx, `UPDATE harmony_machines SET last_contact=CURRENT_TIMESTAMP`)
+			if err != nil {
+				logger.Error("Cannot keepalive ", err)
+			}
+		}
+	}()
+
+	return &reg, nil
+}
+
+func CleanupMachines(ctx context.Context, db *harmonydb.DB) int {
+	ct, err := db.Exec(ctx,
+		`DELETE FROM harmony_machines WHERE last_contact < CURRENT_TIMESTAMP - INTERVAL '1 MILLISECOND' * $1 `,
+		LOOKS_DEAD_TIMEOUT.Milliseconds()) // ms enables unit testing to change timeout.
+	if err != nil {
+		logger.Warn("unable to delete old machines: ", err)
+	}
+	return ct
+}
+
+func (res *Reg) Shutdown() {
+	res.shutdown.Store(true)
+}
+
+func getResources() (res Resources, err error) {
+	b, err := exec.Command(`ps`, `-ef`).CombinedOutput()
+	if err != nil {
+		logger.Warn("Could not safety check for 2+ processes: ", err)
+	} else {
+		found := 0
+		for _, b := range bytes.Split(b, []byte("\n")) {
+			if lotusRE.Match(b) {
+				found++
+			}
+		}
+		if found > 1 {
+			logger.Warn("lotus-provider's defaults are for running alone. Use task maximums or CGroups.")
+		}
+	}
+
+	res = Resources{
+		Cpu: runtime.NumCPU(),
+		Ram: memory.FreeMemory(),
+		Gpu: getGPUDevices(),
+	}
+
+	return res, nil
+}
+
+func DiskFree(path string) (uint64, error) {
+	s := unix.Statfs_t{}
+	err := unix.Statfs(path, &s)
+	if err != nil {
+		return 0, err
+	}
+
+	return s.Bfree * uint64(s.Bsize), nil
+}

lib/harmony/taskhelp/common.go

@@ -0,0 +1,19 @@
+package taskhelp
+
+// SubsetIf returns a subset of the slice for which the predicate is true.
+// It does not allocate memory, but rearranges the list in place.
+// A non-zero list input will always return a non-zero list.
+// The return value is the subset and a boolean indicating whether the subset was sliced.
+func SliceIfFound[T any](slice []T, f func(T) bool) ([]T, bool) {
+	ct := 0
+	for i, v := range slice {
+		if f(v) {
+			slice[ct], slice[i] = slice[i], slice[ct]
+			ct++
+		}
+	}
+	if ct == 0 {
+		return slice, false
+	}
+	return slice[:ct], true
+}

lib/promise/promise.go

@@ -0,0 +1,47 @@
+package promise
+
+import (
+	"context"
+	"sync"
+)
+
+type Promise[T any] struct {
+	val  T
+	done chan struct{}
+	mu   sync.Mutex
+}
+
+func (p *Promise[T]) Set(val T) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	// Set value
+	p.val = val
+
+	// Initialize the done channel if it hasn't been initialized
+	if p.done == nil {
+		p.done = make(chan struct{})
+	}
+
+	// Signal that the value is set
+	close(p.done)
+}
+
+func (p *Promise[T]) Val(ctx context.Context) T {
+	p.mu.Lock()
+	// Initialize the done channel if it hasn't been initialized
+	if p.done == nil {
+		p.done = make(chan struct{})
+	}
+	p.mu.Unlock()
+
+	select {
+	case <-ctx.Done():
+		return *new(T)
+	case <-p.done:
+		p.mu.Lock()
+		val := p.val
+		p.mu.Unlock()
+		return val
+	}
+}

lib/promise/promise_test.go

@@ -0,0 +1,65 @@
+package promise
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestPromiseSet(t *testing.T) {
+	p := &Promise[int]{}
+
+	p.Set(42)
+	if p.val != 42 {
+		t.Fatalf("expected 42, got %v", p.val)
+	}
+}
+
+func TestPromiseVal(t *testing.T) {
+	p := &Promise[int]{}
+
+	p.Set(42)
+
+	ctx := context.Background()
+	val := p.Val(ctx)
+
+	if val != 42 {
+		t.Fatalf("expected 42, got %v", val)
+	}
+}
+
+func TestPromiseValWaitsForSet(t *testing.T) {
+	p := &Promise[int]{}
+	var val int
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		ctx := context.Background()
+		val = p.Val(ctx)
+	}()
+
+	time.Sleep(100 * time.Millisecond) // Give some time for the above goroutine to execute
+	p.Set(42)
+	wg.Wait()
+
+	if val != 42 {
+		t.Fatalf("expected 42, got %v", val)
+	}
+}
+
+func TestPromiseValContextCancel(t *testing.T) {
+	p := &Promise[int]{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // Cancel the context
+
+	val := p.Val(ctx)
+
+	var zeroValue int
+	if val != zeroValue {
+		t.Fatalf("expected zero-value, got %v", val)
+	}
+}