Keep Pass history immutable and replenish atomic with decisions
Fixes the three review findings on 50399b6 (pull request #102):
- A batch now rereads every declined name inside its storing transaction,
so a Pass committed after the batch read its input but before it stores
can no longer let the passed dish into the deck (SQ-01).
- A decision that leaves the deck under five creates the replenish batch
and its job in the decision's own transaction, so a decision is never
recorded without the replenish it calls for; when they cannot be
written, neither is the decision, and a retry decides again (SQ-02).
StartBatch commits the batch with its job the same way, replacing the
mutex and the fail-on-enqueue cleanup. jobs gains Kind.EnqueueTx and
Kind.Wake for this.
- Migration 0012 gives each suggestion the name its dish was suggested
under; declined names for batches and the household context's passed
list read it instead of the dinner's mutable name, so renaming a passed
dinner no longer erases it from dedupe (SQ-03).
Task: b4f2558c-227f-4627-852b-be5aa190242a
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+32
-2
@@ -259,6 +259,21 @@ func (k *Kind[P]) Enqueue(ctx context.Context, payload P) (string, error) {
|
||||
return k.r.enqueue(ctx, k.name, data)
|
||||
}
|
||||
|
||||
// EnqueueTx stores a queued job with payload in tx and returns its id, so
|
||||
// the job exists exactly when what else tx writes does. Once tx has
|
||||
// committed, call Wake so an idle worker runs it.
|
||||
func (k *Kind[P]) EnqueueTx(ctx context.Context, tx *sql.Tx, payload P) (string, error) {
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("jobs: enqueue %s: encode payload: %w", k.name, err)
|
||||
}
|
||||
return createJob(ctx, db.New(tx), k.name, data)
|
||||
}
|
||||
|
||||
// Wake tells an idle worker to look for a job, as Enqueue does itself:
|
||||
// call it after committing the transaction a job was enqueued in.
|
||||
func (k *Kind[P]) Wake() { k.r.wakeOne() }
|
||||
|
||||
// Enqueue stores a queued job of kind with its JSON payload and returns its
|
||||
// id, as Kind.Enqueue does for a kind known at compile time. It returns
|
||||
// ErrUnknownKind for a kind with no registered handler and
|
||||
@@ -280,9 +295,20 @@ func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessa
|
||||
// enqueue stores a queued job of kind with its encoded payload and wakes a
|
||||
// worker.
|
||||
func (r *Runner) enqueue(ctx context.Context, kind string, payload []byte) (string, error) {
|
||||
id, err := createJob(ctx, db.New(r.st.Writer), kind, payload)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
r.wakeOne()
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// createJob stores a queued job of kind with its encoded payload through q
|
||||
// and returns its id.
|
||||
func createJob(ctx context.Context, q *db.Queries, kind string, payload []byte) (string, error) {
|
||||
id := store.NewID()
|
||||
now := store.Now().String()
|
||||
if err := db.New(r.st.Writer).CreateJob(ctx, db.CreateJobParams{
|
||||
if err := q.CreateJob(ctx, db.CreateJobParams{
|
||||
ID: id,
|
||||
Kind: kind,
|
||||
Payload: string(payload),
|
||||
@@ -291,11 +317,15 @@ func (r *Runner) enqueue(ctx context.Context, kind string, payload []byte) (stri
|
||||
}); err != nil {
|
||||
return "", fmt.Errorf("jobs: enqueue %s: %w", kind, err)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// wakeOne gives an idle worker a token to look for a queued job.
|
||||
func (r *Runner) wakeOne() {
|
||||
select {
|
||||
case r.wake <- struct{}{}:
|
||||
default: // every worker already has a token and will look again
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// Start fails every job a previous process left running with
|
||||
|
||||
@@ -187,6 +187,46 @@ func TestJobSucceeds(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnqueueTxStoresJobOnlyWithItsTransaction(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := migratedStore(t)
|
||||
r := newRunner(t, st, 1)
|
||||
count := register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
|
||||
return countResult{Done: p.Steps}, nil
|
||||
})
|
||||
start(t, r)
|
||||
ctx := context.Background()
|
||||
|
||||
var rolledBack string
|
||||
refused := errors.New("refused")
|
||||
if err := st.WithTx(ctx, func(tx *sql.Tx) error {
|
||||
var err error
|
||||
rolledBack, err = count.EnqueueTx(ctx, tx, countPayload{Steps: 1})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return refused
|
||||
}); !errors.Is(err, refused) {
|
||||
t.Fatalf("WithTx = %v, want the refusal", err)
|
||||
}
|
||||
if _, err := r.Get(ctx, rolledBack); !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("job of a rolled-back transaction: Get = %v, want ErrNotFound", err)
|
||||
}
|
||||
|
||||
var id string
|
||||
if err := st.WithTx(ctx, func(tx *sql.Tx) error {
|
||||
var err error
|
||||
id, err = count.EnqueueTx(ctx, tx, countPayload{Steps: 2})
|
||||
return err
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
count.Wake()
|
||||
if job := waitFor(t, r, id, finished); job.Status != StatusSucceeded || string(job.Result) != `{"done":2}` {
|
||||
t.Errorf("job = %+v (result %s), want succeeded with {\"done\":2}", job, job.Result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobFailsWithFailureCode(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := migratedStore(t)
|
||||
|
||||
Reference in New Issue
Block a user