Skip to main content

internal/formation/notifications.go

internal/formation · 119 lines · 8 declarations · source

Declarations​

type Notifications​

type Notifications struct {
	store    NotificationStore
	sender   *notify.Sender
	budget   int
	baseWait time.Duration
}

Notifications is the pass that tells a project its memory moved, and the loop that keeps trying.

Why it is a pass beside formation rather than part of it​

Formation's job is to turn stored turns into memory, and it must not be able to fail because somebody's webhook is down. So this reads the watermark formation left behind, writes down what is owed, and sends on its own schedule. The worst a broken destination can do is fill a queue that an operator can read.

source

type NotificationStore​

type NotificationStore interface {
	Owed(ctx context.Context, scope string, formed, stored int64, parked int) (int, error)
	Claim(ctx context.Context) (pg.Due, bool, error)
	Delivered(ctx context.Context, id string, status int) error
	Failed(ctx context.Context, id string, attempts, budget, status int, reason string, backoff time.Duration) error
}

NotificationStore is what this pass needs of the store, and no more. Named here rather than taken as a concrete type so the pass can be exercised against a store that fails on purpose.

source

const DefaultAttemptBudget​

const DefaultAttemptBudget = 6

DefaultAttemptBudget is how many times a delivery is tried before it is parked.

Six, with doubling from five seconds, is about five minutes of trying. Longer would keep a queue full of news that has stopped being news — the receiver reads the watermark and catches up in one call — and shorter would park a destination that restarted.

source

func NewNotifications​

func NewNotifications(store NotificationStore, sender *notify.Sender) *Notifications

NewNotifications builds the pass.

source

method Notifications.WithBudget​

func (n *Notifications) WithBudget(budget int, baseWait time.Duration) *Notifications

WithBudget sets the attempt budget and the first wait, for a test that cannot spend five minutes.

source

method Notifications.Owe​

func (n *Notifications) Owe(ctx context.Context, scope string, formed, stored int64, parked int) (int, error)

Owe records what a scope's watermark says is worth telling. Cheap and idempotent: the row it would write is usually already there, because most passes have no news.

source

method Notifications.Deliver​

func (n *Notifications) Deliver(ctx context.Context, limit int) (delivered int, attempted int, err error)

Deliver attempts up to `limit` due deliveries and reports how many landed.

Bounded per pass because this shares a worker with formation, and a queue of a thousand failing destinations must not be able to stop memory being formed.

source

method Notifications.wait​

func (n *Notifications) wait(attempts int) time.Duration

wait doubles with each attempt, because the failure that matters most is the one that is not this destination's fault: a receiver restarting fails every delivery at once, and a flat retry would spend the whole queue's budget inside a minute.

source