Fix/persist worker down

README.md

@@ -370,7 +370,7 @@ a spell of running clean.
 `down` is the exception, and the only state you can set by hand
 (`cork worker-down <ip>`). An operator taking a box out of service knows
 something the probes do not, so no probe will lift it: it stands until
-`worker-add`.
+`worker-add`, across corkd restarts too.
 
 Use it before a **termination**, not before a reboot. A reboot needs nothing:
 the box is ejected while it is away and rejoins by itself, which is the whole

cork/database.go

@@ -163,12 +163,16 @@ const schemaQuery string = `
 	-- Docker workers for multi-host instance placement. ip is the private
 	-- orchestration address cork dials; public is the player-facing address
 	-- (IP or hostname) surfaced in instance metadata, '' meaning the private
-	-- IP doubles as public. worker-remove purges the row along with the
-	-- worker's instance records. (DBs created before this scheme carry a
-	-- leftover 'removed' column; it is never read.)
+	-- IP doubles as public. downsince is when worker-down took the worker
+	-- down (unix nanos), NULL until then and again after a worker-add: down
+	-- is the one asserted state, so the one a restart cannot re-derive from a
+	-- probe. worker-remove purges the row along with the worker's instance
+	-- records. (DBs created before this scheme carry a leftover 'removed'
+	-- column; it is never read.)
 	CREATE TABLE IF NOT EXISTS workers (
 		ip TEXT NOT NULL PRIMARY KEY,
-		public TEXT NOT NULL DEFAULT ''
+		public TEXT NOT NULL DEFAULT '',
+		downsince INTEGER
 	);
 
 	-- Schemas every worker holds the images of (see preload.go). Not a
@@ -500,6 +504,22 @@ func (m *Manager) initDatabase() error {
 		}
 	}
 
+	// Migrate older DBs: add workers.downsince. NULL is "not down", and a
+	// worker downed before the column existed came back up at that restart.
+	var downSinceCols int
+	err = db.QueryRow("SELECT COUNT(*) FROM pragma_table_info('workers') WHERE name = 'downsince';").Scan(&downSinceCols)
+	if err != nil {
+		m.log.errorf("could not check workers table schema: %s", err)
+		return err
+	}
+	if downSinceCols == 0 {
+		_, err = db.Exec("ALTER TABLE workers ADD COLUMN downsince INTEGER;")
+		if err != nil {
+			m.log.errorf("could not migrate workers.downsince column: %s", err)
+			return err
+		}
+	}
+
 	// Drop the images.digest column left behind by the digest-based worker
 	// freshness check: content-addressed tags make the stored registry digest
 	// redundant (the start path now only checks tag existence). Plain ALTER ...

cork/database_operations_test.go

@@ -268,6 +268,51 @@ func TestChallengesDifficultyMigration(t *testing.T) {
 	}
 }
 
+// TestWorkersDownSinceMigration verifies that a workers table from before
+// worker-down was persisted gains downsince in place, that its rows read back
+// as not down, and that re-running the migration is a no-op.
+func TestWorkersDownSinceMigration(t *testing.T) {
+	dbFile, err := os.CreateTemp("", "cmgr-downsince-migrate-*.db")
+	if err != nil {
+		t.Fatalf("failed to create temp file: %s", err)
+	}
+	dbFile.Close()
+	t.Cleanup(func() { removeDBFiles(dbFile.Name()) })
+
+	dsn := dbFile.Name() + "?_fk=false&_journal_mode=WAL&_busy_timeout=50&_synchronous=NORMAL"
+	legacy, err := sqlx.Open("sqlite3", dsn)
+	if err != nil {
+		t.Fatalf("failed to open db: %s", err)
+	}
+	// With the leftover 'removed' column that older databases still carry.
+	schema := `
+	CREATE TABLE workers (
+		ip TEXT NOT NULL PRIMARY KEY,
+		public TEXT NOT NULL DEFAULT '',
+		removed INTEGER NOT NULL DEFAULT 0
+	);
+	INSERT INTO workers(ip, public) VALUES ('10.0.0.1', 'worker-a.example');`
+	if _, err := legacy.Exec(schema); err != nil {
+		t.Fatalf("failed to create legacy schema: %s", err)
+	}
+	legacy.Close()
+
+	os.Setenv(DB_ENV, dbFile.Name())
+	defer os.Unsetenv(DB_ENV)
+
+	for _, pass := range []string{"migration", "idempotence"} {
+		mgr := new(Manager)
+		mgr.log = newLogger(DISABLED)
+		if err := mgr.initDatabase(); err != nil {
+			t.Fatalf("initDatabase failed (%s pass): %s", pass, err)
+		}
+		if storedDownSince(t, mgr, "10.0.0.1").Valid {
+			t.Errorf("a worker from before the migration reads back as down after the %s pass", pass)
+		}
+		mgr.db.Close()
+	}
+}
+
 // TestDatabaseBuildLifecycle tests the build database operations
 func TestDatabaseBuildLifecycle(t *testing.T) {
 	mgr := setupTestManager(t)

cork/worker_down_test.go

@@ -0,0 +1,117 @@
+package cork
+
+import (
+	"context"
+	"database/sql"
+	"strings"
+	"testing"
+)
+
+// restarted is another corkd started on m's database, as a restart would
+// leave it.
+func restarted(t *testing.T, m *Manager) *Manager {
+	t.Helper()
+	r := &Manager{log: newLogger(DISABLED), ctx: context.Background(), workerTiming: m.workerTiming, db: m.db}
+	if err := r.initWorkers(); err != nil {
+		t.Fatalf("initWorkers after the restart: %s", err)
+	}
+	t.Cleanup(func() {
+		r.workersMu.Lock()
+		defer r.workersMu.Unlock()
+		for _, w := range r.workers {
+			close(w.done)
+		}
+	})
+	return r
+}
+
+func storedDownSince(t *testing.T, m *Manager, ip string) sql.NullInt64 {
+	t.Helper()
+	var downSince sql.NullInt64
+	if err := m.db.Get(&downSince, "SELECT downsince FROM workers WHERE ip = ?;", ip); err != nil {
+		t.Fatalf("reading back downsince of %s: %s", ip, err)
+	}
+	return downSince
+}
+
+// Down is the one state a restart cannot re-derive from a probe, so it is the
+// one that survives it: the worker comes back down, as of when it was taken
+// down, and out of placement. An ejection is an observation and does not.
+func TestWorkerDownSurvivesARestart(t *testing.T) {
+	const downIP, ejectedIP = "127.0.0.1", "127.0.0.2"
+	m := addWorkerTestManager(t)
+	for _, ip := range []string{downIP, ejectedIP} {
+		if err := m.AddWorker(ip, ""); err != nil {
+			t.Fatalf("worker-add %s: %s", ip, err)
+		}
+	}
+	if err := m.SetWorkerDown(downIP); err != nil {
+		t.Fatalf("worker-down: %s", err)
+	}
+	m.workersMu.RLock()
+	since := m.workers[downIP].since.Load()
+	m.eject(m.workers[ejectedIP], "its daemon did not answer")
+	m.workersMu.RUnlock()
+
+	r := restarted(t, m)
+	down := r.workers[downIP]
+	if got := reachableOf(down); got != workerDown {
+		t.Fatalf("reachable %s after a restart, want down: the restart returned a downed worker to placement", got)
+	}
+	if got := down.since.Load(); got != since {
+		t.Errorf("since %d after a restart, want %d, when the operator took it down", got, since)
+	}
+	if reason := down.reason.Load(); reason == nil || *reason != downReason {
+		t.Errorf("reason %v after a restart, want %q", reason, downReason)
+	}
+	if got := reachableOf(r.workers[ejectedIP]); got == workerDown {
+		t.Fatal("an ejected worker came back down: an observation was persisted as an assertion, and no probe will ever lift it")
+	}
+}
+
+// Marked before the write: a worker-down that cannot persist still takes the
+// worker out of placement for the life of this corkd, and its error says so.
+func TestWorkerDownThatCannotPersistStillTakesTheWorkerDown(t *testing.T) {
+	m := &Manager{log: newLogger(DISABLED), ctx: context.Background()}
+	w := testWorkerConn(workerReachableOk)
+	m.workers = map[string]*workerConn{w.ip: w}
+	m.db = workerDB(t, w.ip)
+	m.db.Close()
+
+	err := m.SetWorkerDown(w.ip)
+	if err == nil {
+		t.Fatal("worker-down reported success on a database it could not write")
+	}
+	if got := reachableOf(w); got != workerDown {
+		t.Fatalf("reachable %s after a worker-down whose write failed, want down", got)
+	}
+	if !strings.Contains(err.Error(), "is down") {
+		t.Errorf("error %q does not say the worker is down, so worker-down reads as having done nothing", err)
+	}
+}
+
+// worker-add lifts the persisted assertion along with the one in memory, or
+// the next restart takes the worker back out of placement.
+func TestWorkerAddLiftsAPersistedDown(t *testing.T) {
+	const ip = "127.0.0.1"
+	m := addWorkerTestManager(t)
+	if err := m.AddWorker(ip, ""); err != nil {
+		t.Fatalf("worker-add: %s", err)
+	}
+	if err := m.SetWorkerDown(ip); err != nil {
+		t.Fatalf("worker-down: %s", err)
+	}
+	if !storedDownSince(t, m, ip).Valid {
+		t.Fatal("worker-down was not persisted")
+	}
+
+	if err := m.AddWorker(ip, ""); err != nil {
+		t.Fatalf("worker-add after worker-down: %s", err)
+	}
+	if storedDownSince(t, m, ip).Valid {
+		t.Fatal("worker-add left worker-down persisted, so the next restart takes the worker out of placement again")
+	}
+	if got := reachableOf(restarted(t, m).workers[ip]); got == workerDown {
+		t.Fatal("a worker brought back with worker-add is down again after a restart")
+	}
+}

cork/worker_probe_test.go

@@ -262,6 +262,7 @@ func TestRecoveryDoesNotResurrectAnAssertedDownWorker(t *testing.T) {
 	w := testWorkerConn(workerUnresponsive)
 	w.done = make(chan struct{})
 	m.workers = map[string]*workerConn{w.ip: w}
+	m.db = workerDB(t, w.ip)
 
 	if err := m.SetWorkerDown(w.ip); err != nil {
 		t.Fatalf("worker-down: %s", err)
@@ -317,9 +318,10 @@ func TestWorkerDownAndRecoveryCannotBothWin(t *testing.T) {
 	timing.pollInterval = 5 * time.Millisecond
 	timing.maxMisses = 1
 
+	db := workerDB(t, testWorkerConn(workerUnresponsive).ip)
 	var swapped, declined int
 	for i := 0; i < 100; i++ {
-		m := &Manager{log: newLogger(DISABLED), ctx: context.Background(), workerTiming: timing}
+		m := &Manager{log: newLogger(DISABLED), ctx: context.Background(), workerTiming: timing, db: db}
 		w := testWorkerConn(workerUnresponsive)
 		w.done = make(chan struct{})
 		m.workers = map[string]*workerConn{w.ip: w}
@@ -381,6 +383,7 @@ func TestWorkerDownLandsOnTheConnThatRecoveryInstalled(t *testing.T) {
 	w := testWorkerConn(workerUnresponsive)
 	w.done = make(chan struct{})
 	m.workers = map[string]*workerConn{w.ip: w}
+	m.db = workerDB(t, w.ip)
 
 	m.recoverWorker(w)
 	fresh := m.workers[w.ip]

cork/worker_reconcile.go

@@ -56,7 +56,7 @@ func (r reconcileResult) String() string {
 // DB-only stop leaves behind while the worker is down or purged
 // (stopInstance, RemoveWorker). They are not merely untidy. A box that
 // rejoins placement, through worker-add or a corkd restart that starts every
-// worker afresh, would otherwise refuse any new instance that draws one of
+// worker but a downed one afresh, would otherwise refuse any new instance that draws one of
 // their published host ports. It therefore runs before the worker's poller
 // starts (runWorker), so nothing is placed on the worker until they are gone.
 //