A receipt can carry an acknowledgement id no worker reported

internal/connector/dispatch_lifecycle_test.go

@@ -2,7 +2,10 @@ package connector
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
+	"os"
+	"path/filepath"
 	"slices"
 	"strings"
 	"testing"
@@ -549,3 +552,101 @@ func TestOneWriteCannotWithdrawAndComplete(t *testing.T) {
 	assert.Nil(t, withdrawn)
 	assert.Equal(t, "exposed", f.rowContext(ctx, t, 1).Delivery)
 }
+
+// An acknowledgement id belongs to the statement that acknowledges. Writing it
+// onto a row that stays exposed would leave an id in the receipt that no worker
+// ever reported, which is the half of "with the acknowledgement or never" that
+// checking only the old row cannot see.
+func TestAnAcknowledgementIDIsNotWrittenWithoutTheAcknowledgement(t *testing.T) {
+	ctx := context.Background()
+	f := newDispatchFixture(t)
+	_, err := f.ledger.db.ExecContext(ctx, `UPDATE task_events SET delivery = 'exposed', exposed_at = 'launch' WHERE event_id = 1`)
+	require.NoError(t, err)
+
+	_, err = f.ledger.db.ExecContext(ctx, `UPDATE task_events SET ack_id = 99 WHERE task_id = ? AND event_id = 1`, f.grant.ID)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "an acknowledgement id is written with the acknowledgement, once")
+	var ackID *int64
+	require.NoError(t, f.ledger.db.QueryRowContext(ctx, `SELECT ack_id FROM task_events WHERE task_id = ? AND event_id = 1`, f.grant.ID).Scan(&ackID))
+	assert.Nil(t, ackID)
+	assert.Equal(t, "exposed", f.rowContext(ctx, t, 1).Delivery)
+}
+
+// And the acknowledgement itself still writes one: the rule narrows what may
+// write an id, not whether Ack can.
+func TestAcknowledgingWritesTheIDItWasGiven(t *testing.T) {
+	ctx := context.Background()
+	f := newDispatchFixture(t)
+	_, err := f.ledger.db.ExecContext(ctx, `UPDATE task_events SET delivery = 'exposed', exposed_at = 'launch' WHERE event_id = 1`)
+	require.NoError(t, err)
+
+	d, err := f.ledger.Dispatch(ctx, f.grant.Token, adapterAgentID)
+	require.NoError(t, err)
+	_, _, err = d.Get(ctx, 1)
+	require.NoError(t, err)
+	ackID := int64(99)
+	_, err = d.Ack(ctx, 1, &ackID)
+	require.NoError(t, err)
+
+	var got *int64
+	require.NoError(t, f.ledger.db.QueryRowContext(ctx, `SELECT ack_id FROM task_events WHERE task_id = ? AND event_id = 1`, f.grant.ID).Scan(&got))
+	require.NotNil(t, got)
+	assert.Equal(t, int64(99), *got)
+	assert.Equal(t, "delivered", f.rowContext(ctx, t, 1).Delivery)
+}
+
+// A ledger born under migration 5 carries that migration's trigger, which
+// read only the row as it was. Editing migration 5 would have left every such
+// ledger with it, because migrate skips what it has already applied — so the
+// replacement is migration 6, and this is the upgrade actually happening.
+func TestAnExistingLedgerGetsTheTighterAcknowledgementRule(t *testing.T) {
+	ctx := context.Background()
+	path := filepath.Join(t.TempDir(), "state", "connector.db")
+	require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700))
+
+	// A ledger as the previous version wrote it: migrations 1 through 5 and
+	// nothing after them.
+	old, err := sql.Open("sqlite", ledgerDSN(path, true))
+	require.NoError(t, err)
+	_, err = old.ExecContext(ctx, `CREATE TABLE schema_migrations (version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL)`)
+	require.NoError(t, err)
+	for i := 0; i < 5; i++ {
+		_, err = old.ExecContext(ctx, migrations[i])
+		require.NoError(t, err, "migration %d", i+1)
+		_, err = old.ExecContext(ctx, `INSERT INTO schema_migrations (version, applied_at) VALUES (?, 'then')`, i+1)
+		require.NoError(t, err)
+	}
+	require.NoError(t, old.Close())
+	// The connector's own open makes the file private; a raw sql.Open does
+	// not, and the privacy check refuses what it finds.
+	for _, name := range []string{path, path + "-wal", path + "-shm"} {
+		if _, err := os.Stat(name); err == nil {
+			require.NoError(t, os.Chmod(name, 0o600))
+		}
+	}
+
+	ledger, err := OpenLedger(path)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = ledger.Close() })
+	version, err := ledger.SchemaVersion(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, len(migrations), version, "the upgrade ran")
+
+	// The same refusal the fresh-ledger test pins, on a ledger that was not
+	// born with it.
+	for _, id := range []int64{1, 2} {
+		seenRecord(t, ledger, id)
+		_, err := ledger.Admission().Commit(ctx, admittedVerdict(id, 0, "recording:10304028989"))
+		require.NoError(t, err)
+	}
+	grant, err := ledger.CreateTask(ctx, []int64{1, 2})
+	require.NoError(t, err)
+	_, err = ledger.db.ExecContext(ctx, `UPDATE task_events SET delivery = 'exposed', exposed_at = 'launch' WHERE event_id = 1`)
+	require.NoError(t, err)
+
+	_, err = ledger.db.ExecContext(ctx, `UPDATE task_events SET ack_id = 99 WHERE task_id = ? AND event_id = 1`, grant.ID)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "an acknowledgement id is written with the acknowledgement, once")
+}

internal/connector/ledger.go

@@ -795,6 +795,24 @@ WHEN (OLD.delivery = 'admitted' AND NEW.delivery IN ('delivered', 'completed'))
 BEGIN
   SELECT RAISE(ABORT, 'a worker acknowledges and completes what it pulled; anything else is the dispatcher settling a completed record');
 END;
+`,
+	// 6. The acknowledgement id settles with the acknowledgement.
+	//
+	// Migration 5 shipped a trigger that read only the row as it was, so a
+	// statement could write the id and leave the row exposed — an id in the
+	// receipt that no worker ever reported. A ledger already at version 5
+	// keeps that trigger, so replacing it is its own migration rather than an
+	// edit to one that has shipped.
+	`
+DROP TRIGGER task_events_acknowledgement_settles_once;
+
+CREATE TRIGGER task_events_acknowledgement_settles_once
+BEFORE UPDATE OF ack_id ON task_events
+WHEN NEW.ack_id IS NOT OLD.ack_id
+ AND (OLD.ack_id IS NOT NULL OR OLD.delivery <> 'exposed' OR NEW.delivery <> 'delivered')
+BEGIN
+  SELECT RAISE(ABORT, 'an acknowledgement id is written with the acknowledgement, once');
+END;
 `,
 }