feat!: hoppersqlite, a pure-Go SQLite driver, and a drivertest suite any engine can pass

.github/dependabot.yml

@@ -3,7 +3,7 @@ version: 2
 # compromised release is likely to have been pulled before it is proposed here.
 updates:
   - package-ecosystem: gomod
-    directories: ["/", "/tools", "/tools/dev", "/hopperotel", "/hopperui"]
+    directories: ["/", "/tools", "/tools/dev", "/hopperotel", "/hopperui", "/driver/hoppersqlite"]
     schedule: { interval: weekly }
     cooldown: { default-days: 7 }
     commit-message: { prefix: "build(deps)" }

.github/workflows/ci.yml

@@ -41,11 +41,13 @@ jobs:
           (cd tools/dev && GOTOOLCHAIN=auto go mod tidy -diff)
           (cd hopperotel && go mod tidy -diff)
           (cd hopperui && go mod tidy -diff)
+          (cd driver/hoppersqlite && go mod tidy -diff)
       - name: Lint
         run: |
           go tool -modfile=tools/go.mod golangci-lint run ./...
           (cd hopperotel && go tool -modfile=../tools/go.mod golangci-lint run ./...)
           (cd hopperui && go tool -modfile=../tools/go.mod golangci-lint run ./...)
+          (cd driver/hoppersqlite && go tool -modfile=../../tools/go.mod golangci-lint run ./...)
 
   # Audits the workflows themselves: unpinned actions, script injection,
   # over-broad tokens, leaked credentials. Any finding fails the job.
@@ -97,3 +99,23 @@ jobs:
           go test -race -cover -timeout 10m ./...
           (cd hopperotel && go test -race -cover -timeout 10m ./...)
           (cd hopperui && go test -race -cover -timeout 10m ./...)
+
+  # hoppersqlite needs no server, so it runs once rather than per Postgres
+  # version, and proves that it and everything it imports build without cgo.
+  test-sqlite:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+        with:
+          go-version-file: go.mod
+      - name: Build without cgo
+        run: make cgo-check
+      - name: Test
+        env:
+          GODEBUG: fips140=only
+        working-directory: driver/hoppersqlite
+        run: go test -race -cover -timeout 10m ./...

.github/workflows/release.yml

@@ -67,17 +67,21 @@ jobs:
       - name: Tag the sub-modules at the same version
         # release-please manages one package, the core module, and rewrites
         # the sub-modules' requirement on it (x-release-please-version in
-        # their go.mod). Their tags are added here so that all three modules
-        # resolve at one version.
+        # their go.mod). Their tags are added here so that every module
+        # resolves at one version. A module in a subdirectory is tagged with
+        # its path, as the go command looks it up.
         env:
           TAG: ${{ needs.release-please.outputs.tag_name }}
         run: |
           git config user.name "github-actions[bot]"
           git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
-          for m in hopperotel hopperui; do
+          modules="hopperotel hopperui driver/hoppersqlite"
+          tags=()
+          for m in $modules; do
             git tag -a "$m/$TAG" "$TAG" -m "$m $TAG"
+            tags+=("$m/$TAG")
           done
-          git push origin "hopperotel/$TAG" "hopperui/$TAG"
+          git push origin "${tags[@]}"
       - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
         with:
           go-version-file: go.mod

AGENTS.md

@@ -11,6 +11,9 @@
   owning client. Changes to the SQL in the claim, finalize, rescue or leader paths need
   a concurrency or chaos test.
 - Always use database time (`now()`), never application time, when comparing job times.
+- A schema change is a new version in `hoppermigrate/migrations` and the same version in
+  `driver/hoppersqlite/migrations`, up and down; for a change SQLite has no use for, its
+  script is a comment saying why. hoppersqlite's tests fail for a version it lacks.
 - Changes to the insert, claim or finalize paths include `hopperbench` results in the
   PR. A regression of more than 10% against `canary` blocks the merge.
 - [docs/PLAN.md](docs/PLAN.md) is the design of record. Change it in the same PR as a

Makefile

@@ -42,6 +42,7 @@ lint: ## Lint Go code
 	$(GOTOOL) golangci-lint run ./...
 	cd hopperotel && go tool -modfile=../tools/go.mod golangci-lint run ./...
 	cd hopperui && go tool -modfile=../tools/go.mod golangci-lint run ./...
+	cd driver/hoppersqlite && go tool -modfile=../../tools/go.mod golangci-lint run ./...
 
 .PHONY: fmt
 fmt: ## Format Go code
@@ -58,7 +59,21 @@ test-modules:
 	go test -race -cover -timeout 10m ./...
 	cd hopperotel && go test -race -cover -timeout 10m ./...
 	cd hopperui && go test -race -cover -timeout 10m ./...
+	cd driver/hoppersqlite && go test -race -cover -timeout 10m ./...
 
 .PHONY: bench
 bench: ## Run the benchmark harness against the test database
 	go run ./cmd/hopperbench -database-url "$(HOPPER_TEST_DATABASE_URL)"
+
+.PHONY: bench-sqlite
+bench-sqlite: ## Run the benchmark harness on a SQLite database in a temporary directory
+	cd driver/hoppersqlite && go run ./cmd/hopperbench-sqlite
+
+.PHONY: cgo-check
+cgo-check: ## Check that hoppersqlite, its tests and every module they import use no cgo
+	@cd driver/hoppersqlite && for target in linux/amd64 linux/386 linux/arm darwin/arm64 windows/amd64 freebsd/amd64 openbsd/amd64 netbsd/amd64; do \
+		export GOOS=$${target%/*} GOARCH=$${target#*/}; \
+		CGO_ENABLED=0 go vet ./... || exit 1; \
+		pkgs=$$(CGO_ENABLED=1 go list -deps -test -f '{{if and .CgoFiles (not .Standard)}}{{.ImportPath}}{{end}}' ./...) || exit 1; \
+		if [ -n "$$pkgs" ]; then echo "cgo is not allowed, but these packages use it on $$target:"; echo "$$pkgs"; exit 1; fi; \
+	done

README.md

@@ -5,7 +5,8 @@
 hopper runs background jobs and delivers messages using the PostgreSQL database
 your application already has. There's no broker to operate and no separate worker
 service: it is a library that runs inside your process and scales out with it.
-Postgres is the first engine, behind a driver interface designed for more.
+Postgres is the first engine, behind a driver interface designed for more; SQLite
+is the second, for an application that runs on one machine.
 
 - **Fast:** set-based claim, finalize and insert (including `COPY`), a live
   table that stays small however much history accumulates, and retention by
@@ -36,7 +37,8 @@ Postgres is the first engine, behind a driver interface designed for more.
 > ordering keys, streams with consumers, a SQL contract for producers in other
 > languages), flow control, batches and workflows, plus the `hopper` CLI,
 > `hoppertest`, `hopperotel`, the `hopperui` web UI and the `drivertest`
-> conformance suite. [docs/PLAN.md](docs/PLAN.md) is the plan of record and
+> conformance suite, and a SQLite driver (`driver/hoppersqlite`, pure Go, a module
+> of its own) that passes it. [docs/PLAN.md](docs/PLAN.md) is the plan of record and
 > [docs/getting-started.md](docs/getting-started.md) the place to start.
 > Until v1.0.0, minor versions may change the API. Feedback is welcome
 > through issues and PRs.

await.go

@@ -50,6 +50,12 @@ type attempt struct {
 	logEnded bool
 }
 
+// String names the attempt, as the owner of its worker's writes
+// (driver.WithOwner).
+func (a *attempt) String() string {
+	return fmt.Sprintf("job %s (%s), attempt %d", a.row.ID, a.row.Kind, a.attempt)
+}
+
 func (a *attempt) markHeldByTx() {
 	a.mu.Lock()
 	a.heldByTx = true

client.go

@@ -35,6 +35,10 @@ type Client[TTx any] struct {
 	// clientID is the lease row ID while running. It changes if the lease is
 	// lost and re-registered.
 	clientID atomic.Int64
+	// leaseFrom is when the current lease was last taken or renewed, as
+	// register and the lease loop saw it: its expiry bounds a renewal's
+	// wait.
+	leaseFrom atomic.Pointer[time.Time]
 
 	mu    sync.Mutex
 	state clientState
@@ -330,6 +334,7 @@ func (c *Client[TTx]) declare(ctx context.Context) error {
 }
 
 func (c *Client[TTx]) register(ctx context.Context) (int64, error) {
+	start := time.Now()
 	info, _ := json.Marshal(map[string]any{
 		"pid":    os.Getpid(),
 		"queues": slices.Sorted(maps.Keys(c.cfg.Queues)),
@@ -342,6 +347,7 @@ func (c *Client[TTx]) register(ctx context.Context) (int64, error) {
 	if err != nil {
 		return 0, fmt.Errorf("hopper: register client: %w", err)
 	}
+	c.leaseFrom.Store(&start)
 	return id, nil
 }
 
@@ -494,6 +500,13 @@ func (c *Client[TTx]) wakeQueue(queue string) {
 // ID. Results still in flight are submitted anyway; the finalize statement
 // rejects any whose job was rescued meanwhile and applies the rest, which is
 // equivalent to a rescue.
+//
+// A renewal waits until the lease it renews would expire, and at least one
+// renewal interval: on an engine that runs one writer at a time, a
+// transaction that holds the database holds every renewal up, and a renewal
+// that gives up is reported, with what the engine says of who holds the
+// lock, and tried again, rather than left waiting past the lease without a
+// word.
 func (c *Client[TTx]) leaseLoop(ctx context.Context) {
 	ticker := time.NewTicker(c.tuning.leaseRenew)
 	defer ticker.Stop()
@@ -504,17 +517,21 @@ func (c *Client[TTx]) leaseLoop(ctx context.Context) {
 		case <-ticker.C:
 		}
 		id := c.clientID.Load()
-		res, err := c.exec.ClientRenew(ctx, driver.ClientRenewParams{ClientID: id, TTL: c.tuning.leaseTTL})
+		start := time.Now()
+		rctx, cancel := context.WithTimeout(ctx, c.renewalWait())
+		res, err := c.exec.ClientRenew(rctx, driver.ClientRenewParams{ClientID: id, TTL: c.tuning.leaseTTL})
+		cancel()
 		if err != nil {
 			if ctx.Err() == nil {
 				c.logger.WarnContext(ctx, "hopper: renew client lease", "client_id", id, "error", err)
 			}
 			continue
 		}
 		if !res.Renewed {
-			c.fence(ctx, id)
+			c.fence(ctx, id) // which registers again, and so records the new lease
 			continue
 		}
+		c.leaseFrom.Store(&start)
 		// Control state rides along with every renewal, in case a
 		// notification was missed.
 		for _, jobID := range res.CancelRequested {
@@ -525,6 +542,16 @@ func (c *Client[TTx]) leaseLoop(ctx context.Context) {
 	}
 }
 
+// renewalWait is how long a renewal may wait: until the lease would expire,
+// and at least one renewal interval.
+func (c *Client[TTx]) renewalWait() time.Duration {
+	wait := c.tuning.leaseRenew
+	if from := c.leaseFrom.Load(); from != nil {
+		wait = max(wait, time.Until(from.Add(c.tuning.leaseTTL)))
+	}
+	return wait
+}
+
 // limitedQueues reads which queues have limits.
 func (c *Client[TTx]) limitedQueues(ctx context.Context) ([]string, error) {
 	queues, err := c.exec.QueueList(ctx)

complete.go

@@ -30,13 +30,22 @@ var ErrJobFenced = errors.New("hopper: job is no longer running in this attempt"
 // whatever the worker returned, and the job.completed event fires. If it did
 // not, the attempt fails with the worker's error, or with a failure if the
 // worker returned nil, and the job is retried. A worker that returns before
-// tx ends holds up only its own job's result, until tx ends.
+// tx ends holds up only its own job's result, until tx ends, on an engine
+// that locks rows.
 //
 // Call it last, just before the commit. From CompleteTx until tx ends, tx
 // holds the job's row and those of workflow dependents it promotes or
 // cancels, and holds off the creation and drop of history partitions. The
 // job's siblings in its batch finalize without waiting for it.
 //
+// On an engine that runs one writer at a time
+// (driver.Capabilities.SingleWriter, SQLite), tx holds the whole database
+// for writes from its first statement until it ends: every other write
+// waits for it, the results of other jobs and lease renewals included, and
+// a write of the worker's own on the pool fails rather than wait for tx,
+// provided tx was begun with the worker's ctx (or one derived from it),
+// which names the job's attempt.
+//
 //	func (w *Worker) Work(ctx context.Context, job *hopper.Job[Args]) error {
 //		tx, err := w.pool.Begin(ctx)
 //		if err != nil {

control.go

@@ -47,7 +47,8 @@ func (c *Client[TTx]) JobCancel(ctx context.Context, id JobID) (*JobRow, error)
 //
 // Until tx ends it holds the job's row, so a running job that finishes
 // meanwhile has its result wait for tx; its client's other results do not.
-// Call it late in tx, as CompleteTx.
+// Call it late in tx, as CompleteTx. On an engine that runs one writer at a
+// time, tx holds every write in the database up until it ends.
 func (c *Client[TTx]) JobCancelTx(ctx context.Context, tx TTx, id JobID) (*JobRow, error) {
 	return c.driver.UnwrapTx(tx).JobCancel(ctx, id)
 }
@@ -114,7 +115,8 @@ func (c *Client[TTx]) JobDelete(ctx context.Context, id JobID) (*JobRow, error)
 
 // JobDeleteTx is JobDelete inside the caller's transaction. Until tx ends
 // it holds the job's row and, for a waiting job, those of the workflow
-// dependents it cancels or promotes.
+// dependents it cancels or promotes; on an engine that runs one writer at a
+// time, every write in the database.
 func (c *Client[TTx]) JobDeleteTx(ctx context.Context, tx TTx, id JobID) (*JobRow, error) {
 	return c.driver.UnwrapTx(tx).JobDelete(ctx, id)
 }
@@ -144,7 +146,8 @@ func (c *Client[TTx]) JobRetryOrderingKey(ctx context.Context, queue, orderingKe
 //
 // Until tx ends it holds the rows of the jobs it woke and of the key's
 // running job, whose result waits for tx; its client's other results do
-// not. Call it late in tx, as CompleteTx.
+// not. Call it late in tx, as CompleteTx. On an engine that runs one writer
+// at a time, tx holds every write in the database up until it ends.
 func (c *Client[TTx]) JobRetryOrderingKeyTx(ctx context.Context, tx TTx, queue, orderingKey string) (int, error) {
 	queue = cmp.Or(queue, QueueDefault)
 	exec := c.driver.UnwrapTx(tx)