feat: finalizers

.markdownlint.json

@@ -0,0 +1,5 @@
+{
+  "MD013": false,
+  "MD060": false,
+  "MD029": false
+}

README.md

@@ -16,6 +16,7 @@ A Go library for delayed task processing and state reconciliation.
     - [Options](#options)
     - [Delay Strategies](#delay-strategies)
     - [Groups](#groups)
+    - [Finalizers](#finalizers)
   - [Configuration](#configuration)
   - [Storage](#storage)
     - [In-Memory](#in-memory)
@@ -32,6 +33,7 @@ A Go library for delayed task processing and state reconciliation.
 - **Retry Strategies**: Fixed delays or exponential backoff
 - **Tubes**: Route tickets to separate queues
 - **Groups**: Track batches of related tickets and query their collective progress
+- **Finalizers**: Run a ticket automatically after all members of a group have finished
 - **Automatic Expiration**: Cleanup of completed/expired tickets
 - **Concurrent Processing**: Configurable worker pools
 - **Prometheus Metrics**: Per-type and per-tube counters, processing duration and queue wait histograms
@@ -160,6 +162,7 @@ kh.Cancel(ctx, id, lymbo.WithKeep(), lymbo.WithErrorReason("cancelled by user"))
 | `WithGroup(id)` | Assign or transfer ticket to a group |
 | `WithUpdate(fn)` | Custom ticket modification (executed last) |
 | `WithResetAttempts()` | Reset attempt counter |
+| `AfterGroup(id)` | Register ticket as finalizer for a group (blocked until all current members finish) |
 
 ### Delay Strategies
 
@@ -222,6 +225,34 @@ kh.Retry(ctx, t.ID,
 
 Tickets submitted without `WithGroup` are ungrouped and never appear in any group query.
 
+### Finalizers
+
+A finalizer is a ticket blocked until every pending member of a group has finished. Submit group
+members first, then submit the finalizer with `AfterGroup`:
+
+```go
+groupID := "order-42-notifications"
+
+for _, userID := range recipients {
+    ticket, _ := lymbo.NewTicket(lymbo.TicketId(uuid.NewString()), "send-notification")
+    kh.Put(ctx, *ticket, lymbo.WithGroup(groupID), lymbo.WithPayload(userID))
+}
+
+finalizer, _ := lymbo.NewTicket(lymbo.TicketId(uuid.NewString()), "notifications-complete")
+kh.Put(ctx, *finalizer, lymbo.AfterGroup(groupID))
+```
+
+At submission time, the store snapshots all currently pending group members as dependencies.
+The finalizer is invisible to workers until every captured dependency reaches a terminal state.
+Tickets added to the group after the finalizer is submitted are not captured.
+
+Notes:
+
+- Empty group (no pending members) → finalizer is immediately eligible.
+- `WithDelay` on the finalizer is evaluated independently of dependencies.
+- `AfterGroup("g")` + `WithGroup("g")` on the same `Put` returns `ErrFinalizerInGroup`.
+- Re-submitting a finalizer with the same ID is a no-op.
+
 ## Configuration
 
 ```go
@@ -292,6 +323,7 @@ type Store interface {
     PollPending(context.Context, PollRequest) (PollResult, error)
     ExpireTickets(ctx context.Context, limit int, now time.Time) ([]TransitionInfo, error)
     CountPendingInGroup(ctx context.Context, groupID string) (int, error)
+    PutAfterGroup(ctx context.Context, ticket Ticket, groupID string) error
 }
 ```
 

errors.go

@@ -10,4 +10,5 @@ var (
 	ErrTicketIDInvalid         = errors.New("ticket ID is invalid")
 	ErrTicketNotFound          = errors.New("ticket not found")
 	ErrInvalidStatusTransition = errors.New("invalid status transition")
+	ErrFinalizerInGroup        = errors.New("finalizer must not be a member of the group it finalizes")
 )

kharon.go

@@ -257,9 +257,15 @@ func (k *Kharon) Put(ctx context.Context, t Ticket, opts ...Option) error {
 	if err != nil {
 		return err
 	}
+	if o.afterGroup != nil && o.groupID != nil && *o.afterGroup == *o.groupID {
+		return ErrFinalizerInGroup
+	}
 	if err := beforeUpdate(ctx, &t, o); err != nil {
 		return err
 	}
+	if o.afterGroup != nil {
+		return k.store.PutAfterGroup(ctx, t, *o.afterGroup)
+	}
 	return k.store.Put(ctx, t)
 }
 

observable.go

@@ -26,6 +26,14 @@ func (o *observableStore) Put(ctx context.Context, t Ticket) error {
 	return nil
 }
 
+func (o *observableStore) PutAfterGroup(ctx context.Context, t Ticket, groupID string) error {
+	if err := o.Store.PutAfterGroup(ctx, t, groupID); err != nil {
+		return err
+	}
+	o.stats.ByKey(t.Type, t.Tube.String()).Inc(stats.Add)
+	return nil
+}
+
 func (o *observableStore) ExpireTickets(ctx context.Context, limit int, now time.Time) ([]TransitionInfo, error) {
 	infos, err := o.Store.ExpireTickets(ctx, limit, now)
 	if err != nil {

options.go

@@ -59,6 +59,9 @@ type Opts struct {
 	// groupID associates the ticket with a group (effective only on Put).
 	groupID *string
 
+	// afterGroup wires the ticket as a finalizer for the given group (effective only on Put).
+	afterGroup *string
+
 	// update allows custom modification of the ticket.
 	update func(ctx context.Context, t *Ticket) error
 }
@@ -156,6 +159,16 @@ func WithGroup(groupID string) Option {
 	}
 }
 
+// AfterGroup wires the ticket as a finalizer for groupID.
+// The finalizer is blocked until all currently pending group members reach a terminal state.
+// Only effective when passed to Put.
+func AfterGroup(groupID string) Option {
+	return func(o *Opts) error {
+		o.afterGroup = &groupID
+		return nil
+	}
+}
+
 // WithTube sets the tube to transfer the ticket to.
 func WithTube(tube Tube) Option {
 	return func(o *Opts) error {

store.go

@@ -78,6 +78,12 @@ type Store interface {
 
 	// CountPendingInGroup returns the number of pending tickets with the given group ID.
 	CountPendingInGroup(ctx context.Context, groupID string) (int, error)
+
+	// PutAfterGroup atomically inserts the ticket and wires it as a finalizer for groupID.
+	// It scans all currently pending group members and creates dep rows blocking the finalizer.
+	// Returns nil if the ticket ID already exists (idempotent re-submission).
+	// Returns ErrFinalizerInGroup if the ticket belongs to the same group it finalizes.
+	PutAfterGroup(ctx context.Context, ticket Ticket, groupID string) error
 }
 
 // TransitionInfo describes a ticket that was affected by a store operation.

store/memory/memory.go

@@ -16,6 +16,7 @@ import (
 type Store struct {
 	mu   sync.RWMutex
 	data map[lymbo.TicketId]lymbo.Ticket
+	deps map[lymbo.TicketId]map[lymbo.TicketId]struct{} // deps[ticket_id] = set of blocked_by IDs
 }
 
 // Ensure Store implements lymbo.Store interface.
@@ -25,6 +26,7 @@ var _ lymbo.Store = (*Store)(nil)
 func NewStore() *Store {
 	return &Store{
 		data: make(map[lymbo.TicketId]lymbo.Ticket),
+		deps: make(map[lymbo.TicketId]map[lymbo.TicketId]struct{}),
 	}
 }
 
@@ -61,6 +63,49 @@ func (m *Store) Delete(_ context.Context, id lymbo.TicketId) error {
 	defer m.mu.Unlock()
 
 	delete(m.data, id)
+	m.removeDep(id)
+	return nil
+}
+
+// removeDep removes id from all dep sets and deletes its own dep set.
+// Must be called with m.mu held.
+func (m *Store) removeDep(id lymbo.TicketId) {
+	for _, depSet := range m.deps {
+		delete(depSet, id)
+	}
+	delete(m.deps, id)
+}
+
+// PutAfterGroup atomically inserts the ticket and wires it as a finalizer for groupID.
+func (m *Store) PutAfterGroup(_ context.Context, t lymbo.Ticket, groupID string) error {
+	if t.ID == "" {
+		return lymbo.ErrTicketIDEmpty
+	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	// REQ 12: no-op if ticket already exists
+	if _, exists := m.data[t.ID]; exists {
+		return nil
+	}
+
+	t.Status = status.Pending
+	m.data[t.ID] = t
+
+	// REQ 6: scan pending group members, insert dep rows
+	for id, member := range m.data {
+		if id == t.ID {
+			continue
+		}
+		if member.GroupId != nil && *member.GroupId == groupID && member.Status == status.Pending {
+			if m.deps[t.ID] == nil {
+				m.deps[t.ID] = make(map[lymbo.TicketId]struct{})
+			}
+			m.deps[t.ID][id] = struct{}{}
+		}
+	}
+
 	return nil
 }
 
@@ -73,6 +118,7 @@ func (m *Store) DeleteBatch(_ context.Context, ids []lymbo.TicketId) ([]lymbo.Tr
 		if t, ok := m.data[id]; ok {
 			infos = append(infos, lymbo.TransitionInfo{Id: t.ID, Type: t.Type, Tube: t.Tube, Status: t.Status})
 			delete(m.data, id)
+			m.removeDep(id)
 		}
 	}
 	return infos, nil
@@ -121,6 +167,13 @@ func (m *Store) UpdateBatch(ctx context.Context, updates []lymbo.UpdateSet) ([]l
 		updateOne(&t, us)
 		m.data[t.ID] = t
 		infos = append(infos, lymbo.TransitionInfo{Id: t.ID, Type: t.Type, Tube: t.Tube, Status: t.Status})
+
+		// REQ 3: remove from all dep sets when ticket reaches terminal state
+		if t.Status == status.Done || t.Status == status.Failed || t.Status == status.Cancelled {
+			for _, depSet := range m.deps {
+				delete(depSet, t.ID)
+			}
+		}
 	}
 
 	return infos, nil
@@ -180,6 +233,11 @@ func (m *Store) PollPending(_ context.Context, req lymbo.PollRequest) (lymbo.Pol
 			continue
 		}
 
+		// REQ 1: blocked tickets must not be returned
+		if len(m.deps[t.ID]) > 0 {
+			continue
+		}
+
 		ready = append(ready, t)
 	}
 

store/postgres/postgres.go

@@ -206,23 +206,17 @@ func (r *Tickets) DeleteBatch(ctx context.Context, ids []lymbo.TicketId) ([]lymb
 		ticketUUIDs = append(ticketUUIDs, ticketUUID)
 	}
 
-	batch := &pgx.Batch{}
-	for _, ticketUUID := range ticketUUIDs {
-		batch.Queue(r.queries.delete, ticketUUID)
+	rows, err := r.db.Query(ctx, r.queries.deleteBatch, ticketUUIDs)
+	if err != nil {
+		return nil, err
 	}
-
-	br := r.db.SendBatch(ctx, batch)
-	defer br.Close() //nolint:errcheck
+	defer rows.Close()
 
 	var infos []lymbo.TransitionInfo
-	for range ticketUUIDs {
+	for rows.Next() {
 		var id uuid.UUID
 		var ticketType, tube string
-		err := br.QueryRow().Scan(&id, &ticketType, &tube)
-		if err != nil {
-			if errors.Is(err, pgx.ErrNoRows) {
-				continue
-			}
+		if err := rows.Scan(&id, &ticketType, &tube); err != nil {
 			return infos, err
 		}
 		infos = append(infos, lymbo.TransitionInfo{
@@ -231,7 +225,7 @@ func (r *Tickets) DeleteBatch(ctx context.Context, ids []lymbo.TicketId) ([]lymb
 			Tube: lymbo.Tube(tube),
 		})
 	}
-	return infos, nil
+	return infos, rows.Err()
 }
 
 func (r *Tickets) Update(ctx context.Context, id lymbo.TicketId, fn lymbo.UpdateFunc) error {
@@ -337,6 +331,9 @@ func (r *Tickets) UpdateBatch(ctx context.Context, updates []lymbo.UpdateSet) ([
 	if len(updates) == 0 {
 		return nil, nil
 	}
+
+	// Dep rows for terminal transitions are deleted atomically inside the update/backoff
+	// query templates via a CTE (REQ 3). No explicit transaction needed here.
 	batch := &pgx.Batch{}
 
 	for _, us := range updates {
@@ -348,7 +345,7 @@ func (r *Tickets) UpdateBatch(ctx context.Context, updates []lymbo.UpdateSet) ([
 		var q string
 		req := []any{ticketUUID}
 
-		// status as  $2
+		// status as $2
 		switch {
 		case us.Status != nil:
 			req = append(req, sql.NullString{String: us.Status.String(), Valid: true})
@@ -438,6 +435,9 @@ func (r *Tickets) UpdateBatch(ctx context.Context, updates []lymbo.UpdateSet) ([
 			Status: s,
 		})
 	}
+	if err := br.Close(); err != nil {
+		return infos, err
+	}
 
 	return infos, nil
 }
@@ -578,6 +578,39 @@ func (r *Tickets) PollPending(ctx context.Context, req lymbo.PollRequest) (lymbo
 	}, nil
 }
 
+func (r *Tickets) PutAfterGroup(ctx context.Context, ticket lymbo.Ticket, groupID string) error {
+	ticketUUID, err := uuid.Parse(ticket.ID.String())
+	if err != nil {
+		return lymbo.ErrTicketIDInvalid
+	}
+
+	var mtime pgtype.Timestamptz
+	if ticket.Mtime != nil {
+		mtime = pgtype.Timestamptz{Time: *ticket.Mtime, Valid: true}
+	}
+	tube := ticket.Tube.String()
+	if tube == "" {
+		tube = "default"
+	}
+
+	_, err = r.db.Exec(ctx, r.queries.putAfterGroup,
+		ticketUUID,             // $1  id
+		ticket.Status.String(), // $2  status
+		pgtype.Timestamptz{Time: ticket.Runat, Valid: true}, // $3  runat
+		int16(ticket.Nice), // $4  nice
+		ticket.Type,        // $5  type
+		pgtype.Timestamptz{Time: ticket.Ctime, Valid: true}, // $6  ctime
+		mtime,                  // $7  mtime
+		int32(ticket.Attempts), // $8  attempts
+		ticket.Payload,         // $9  payload
+		ticket.ErrorReason,     // $10 error_reason
+		tube,                   // $11 tube
+		ticket.GroupId,         // $12 group_id (finalizer's own group, may be nil)
+		groupID,                // $13 group to wait for
+	)
+	return err
+}
+
 func (r *Tickets) CountPendingInGroup(ctx context.Context, groupID string) (int, error) {
 	var count int
 	err := r.db.QueryRow(ctx, r.queries.countPendingInGroup, groupID).Scan(&count)