Open SQLite with WAL and bound delivery re-dispatch by ownership (closes #256)

internal/delivery/inflight.go

@@ -0,0 +1,110 @@
+package delivery
+
+import "sync"
+
+// inflightSet records which deliveries the engine currently owns.
+//
+// A delivery is owned from the moment a task for it is handed to a
+// channel or to a retry timer until the engine has no further plan for
+// it in memory. Restart recovery and both arms of the periodic sweep
+// re-dispatch only deliveries the set does not hold, which is what
+// makes them exact rather than a guess about how long a row has sat at
+// pending.
+//
+// This replaces reasoning from timestamps. A delivery's row says
+// pending from creation until its outcome is written, which covers
+// four different situations — never dispatched, waiting in a channel,
+// being attempted right now, and genuinely stranded — and no column
+// distinguishes them. Only the engine knows which, and it knows
+// exactly. `deliveryChannelSize` is 10000 against 10 workers, so a
+// perfectly healthy delivery can wait far longer than any age bound
+// worth setting before its attempt even begins; an age bound alone
+// re-sends it. See
+// https://git.eeqj.de/sneak/webhooker/issues/256.
+//
+// In-memory state is sufficient because a data directory admits one
+// process: internal/datadir takes an flock on it at startup and a
+// second instance refuses to run. Deliveries owned by a process that
+// died are not in any successor's set, and restart recovery is what
+// picks those up.
+//
+// References are counted rather than held as a plain set because
+// ownership outlives the worker that took it. A target that schedules
+// a retry from inside Deliver adds a reference while the worker still
+// holds one, so the delivery stays owned across the gap between the
+// worker returning and the timer firing — the window in which a sweep
+// would otherwise find the row at retrying and send it again.
+//
+// The zero value is ready to use, and the Engine holds one by value.
+// That is deliberate: an engine built by a constructor that forgot to
+// initialise this would not refuse to re-dispatch anything, and the
+// symptom would be duplicate deliveries rather than a failure anybody
+// notices.
+type inflightSet struct {
+	mu  sync.Mutex
+	ids map[string]int
+}
+
+// retain adds a reference to a delivery the caller already knows the
+// engine owns, so that ownership survives the current holder letting
+// go. It cannot fail.
+func (s *inflightSet) retain(deliveryID string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.ids == nil {
+		s.ids = make(map[string]int)
+	}
+
+	s.ids[deliveryID]++
+}
+
+// retainIdle takes the first reference on a delivery, and reports
+// whether it got it. It fails when the engine already owns the
+// delivery, which is what makes two claimants — restart recovery and
+// the sweep run concurrently, or two sweep arms — mutually exclusive
+// rather than merely atomic.
+func (s *inflightSet) retainIdle(deliveryID string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.ids[deliveryID] > 0 {
+		return false
+	}
+
+	if s.ids == nil {
+		s.ids = make(map[string]int)
+	}
+
+	s.ids[deliveryID] = 1
+
+	return true
+}
+
+// release drops one reference. The delivery becomes eligible for
+// re-dispatch again once the last one goes.
+func (s *inflightSet) release(deliveryID string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	n := s.ids[deliveryID] - 1
+	if n <= 0 {
+		delete(s.ids, deliveryID)
+
+		return
+	}
+
+	s.ids[deliveryID] = n
+}
+
+// held reports how many deliveries the engine currently owns. It
+// exists so a test can assert that ownership is released rather than
+// leaked: a reference that is never dropped hides its delivery from
+// every sweep for the life of the process, which is the one way this
+// mechanism can fail silently.
+func (s *inflightSet) held() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return len(s.ids)
+}