You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
451 lines
12 KiB
451 lines
12 KiB
package alarmdelivery
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"tera-clone/services/tera-api/internal/storage/atomicfile"
|
|
)
|
|
|
|
const defaultMaxJobs = 256
|
|
|
|
const maxStateBytes = 16 * 1024 * 1024
|
|
|
|
type DispatcherOptions struct {
|
|
StatePath string
|
|
MaxJobs int
|
|
Clock func() time.Time
|
|
PollInterval time.Duration
|
|
}
|
|
|
|
type Dispatcher struct {
|
|
statePath string
|
|
maxJobs int
|
|
clock func() time.Time
|
|
pollInterval time.Duration
|
|
transports map[string]Transport
|
|
jobs map[string]deliveryJob
|
|
sequence int
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
wake chan struct{}
|
|
wg sync.WaitGroup
|
|
closeOnce sync.Once
|
|
mu sync.Mutex
|
|
}
|
|
|
|
type deliveryJob struct {
|
|
DeliveryID string `json:"delivery_id"`
|
|
State string `json:"state"`
|
|
Request Request `json:"request"`
|
|
Attempts int `json:"attempts"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
AcceptedAt time.Time `json:"accepted_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
NextAttempt time.Time `json:"next_attempt"`
|
|
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
|
}
|
|
|
|
type dispatcherSnapshot struct {
|
|
FormatVersion int `json:"format_version"`
|
|
Sequence int `json:"sequence"`
|
|
Jobs map[string]deliveryJob `json:"jobs"`
|
|
}
|
|
|
|
func NewDispatcher(transports map[string]Transport, options DispatcherOptions) (*Dispatcher, error) {
|
|
if options.StatePath == "" || !filepath.IsAbs(options.StatePath) {
|
|
return nil, errors.New("alarm delivery state path must be absolute")
|
|
}
|
|
if options.MaxJobs == 0 {
|
|
options.MaxJobs = defaultMaxJobs
|
|
}
|
|
if options.MaxJobs < 1 || options.MaxJobs > 4096 {
|
|
return nil, errors.New("alarm delivery max jobs must be between 1 and 4096")
|
|
}
|
|
if options.Clock == nil {
|
|
options.Clock = time.Now
|
|
}
|
|
if options.PollInterval <= 0 {
|
|
options.PollInterval = 250 * time.Millisecond
|
|
}
|
|
if options.PollInterval < 10*time.Millisecond || options.PollInterval > time.Second {
|
|
return nil, errors.New("alarm delivery poll interval must be between 10ms and 1s")
|
|
}
|
|
if err := ensureStateDirectory(filepath.Dir(options.StatePath)); err != nil {
|
|
return nil, err
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
dispatcher := &Dispatcher{
|
|
statePath: filepath.Clean(options.StatePath), maxJobs: options.MaxJobs,
|
|
clock: options.Clock, pollInterval: options.PollInterval,
|
|
transports: make(map[string]Transport), jobs: make(map[string]deliveryJob),
|
|
ctx: ctx, cancel: cancel, wake: make(chan struct{}, 1),
|
|
}
|
|
for channel, transport := range transports {
|
|
if transport != nil {
|
|
dispatcher.transports[channel] = transport
|
|
}
|
|
}
|
|
if err := dispatcher.load(); err != nil {
|
|
cancel()
|
|
return nil, err
|
|
}
|
|
dispatcher.wg.Add(1)
|
|
go dispatcher.worker()
|
|
dispatcher.notify()
|
|
return dispatcher, nil
|
|
}
|
|
|
|
func (d *Dispatcher) Deliver(ctx context.Context, request Request) (Result, error) {
|
|
request, err := NormalizeRequest(request)
|
|
if err != nil {
|
|
return Result{}, err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return Result{}, err
|
|
}
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
if d.ctx.Err() != nil {
|
|
return Result{}, &UnavailableError{Reason: "tera-alarmd is stopping"}
|
|
}
|
|
d.trimTerminalLocked()
|
|
active := 0
|
|
for _, job := range d.jobs {
|
|
if !terminalDeliveryState(job.State) {
|
|
active++
|
|
}
|
|
}
|
|
queueLimit := request.QueueLimit
|
|
if queueLimit > d.maxJobs {
|
|
queueLimit = d.maxJobs
|
|
}
|
|
if active >= queueLimit || len(d.jobs) >= d.maxJobs {
|
|
return Result{}, errors.New("alarm delivery queue is full")
|
|
}
|
|
now := d.clock().UTC()
|
|
d.sequence++
|
|
deliveryID := fmt.Sprintf("alarm_delivery_%s_%06d", now.Format("20060102"), d.sequence)
|
|
job := deliveryJob{
|
|
DeliveryID: deliveryID, State: "queued", Request: request,
|
|
AcceptedAt: now, UpdatedAt: now, NextAttempt: now,
|
|
}
|
|
d.jobs[deliveryID] = job
|
|
if err := d.persistLocked(); err != nil {
|
|
delete(d.jobs, deliveryID)
|
|
d.sequence--
|
|
return Result{}, err
|
|
}
|
|
d.notify()
|
|
return resultFromJob(job), nil
|
|
}
|
|
|
|
func (d *Dispatcher) Status(ctx context.Context, deliveryID string) (Result, error) {
|
|
if err := validateDeliveryID(deliveryID); err != nil {
|
|
return Result{}, err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return Result{}, err
|
|
}
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
job, ok := d.jobs[deliveryID]
|
|
if !ok {
|
|
return Result{}, errors.New("alarm delivery not found")
|
|
}
|
|
return resultFromJob(job), nil
|
|
}
|
|
|
|
func (d *Dispatcher) Close() {
|
|
d.closeOnce.Do(func() {
|
|
d.cancel()
|
|
d.wg.Wait()
|
|
})
|
|
}
|
|
|
|
func (d *Dispatcher) worker() {
|
|
defer d.wg.Done()
|
|
ticker := time.NewTicker(d.pollInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-d.ctx.Done():
|
|
return
|
|
case <-d.wake:
|
|
d.processDue()
|
|
case <-ticker.C:
|
|
d.processDue()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (d *Dispatcher) processDue() {
|
|
for {
|
|
d.mu.Lock()
|
|
now := d.clock().UTC()
|
|
var selected deliveryJob
|
|
found := false
|
|
for _, job := range d.jobs {
|
|
if (job.State != "queued" && job.State != "retry_pending") || job.NextAttempt.After(now) {
|
|
continue
|
|
}
|
|
if !found || job.NextAttempt.Before(selected.NextAttempt) || (job.NextAttempt.Equal(selected.NextAttempt) && job.DeliveryID < selected.DeliveryID) {
|
|
selected = job
|
|
found = true
|
|
}
|
|
}
|
|
if !found {
|
|
d.mu.Unlock()
|
|
return
|
|
}
|
|
selected.State = "running"
|
|
selected.UpdatedAt = now
|
|
d.jobs[selected.DeliveryID] = selected
|
|
if err := d.persistLocked(); err != nil {
|
|
selected.State = "queued"
|
|
d.jobs[selected.DeliveryID] = selected
|
|
d.mu.Unlock()
|
|
return
|
|
}
|
|
d.mu.Unlock()
|
|
|
|
attemptCtx, cancel := context.WithTimeout(d.ctx, 10*time.Second)
|
|
err := d.send(attemptCtx, selected)
|
|
cancel()
|
|
|
|
d.mu.Lock()
|
|
job, ok := d.jobs[selected.DeliveryID]
|
|
if !ok {
|
|
d.mu.Unlock()
|
|
continue
|
|
}
|
|
now = d.clock().UTC()
|
|
job.Attempts++
|
|
job.UpdatedAt = now
|
|
if err == nil {
|
|
job.State = "delivered"
|
|
job.LastError = ""
|
|
job.CompletedAt = timePointer(now)
|
|
} else {
|
|
job.LastError = boundedDeliveryError(err)
|
|
if job.Attempts <= job.Request.RetryCount {
|
|
job.State = "retry_pending"
|
|
job.NextAttempt = now.Add(time.Duration(job.Request.RetryIntervalSeconds) * time.Second)
|
|
} else {
|
|
job.State = "failed"
|
|
job.CompletedAt = timePointer(now)
|
|
}
|
|
}
|
|
d.jobs[job.DeliveryID] = job
|
|
d.trimTerminalLocked()
|
|
_ = d.persistLocked()
|
|
d.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (d *Dispatcher) send(ctx context.Context, job deliveryJob) error {
|
|
for _, channel := range job.Request.Channels {
|
|
transport := d.transports[channel]
|
|
if transport == nil {
|
|
return fmt.Errorf("alarm delivery %s transport is not configured", channel)
|
|
}
|
|
if err := transport.Send(ctx, Attempt{DeliveryID: job.DeliveryID, Channel: channel, Request: job.Request}); err != nil {
|
|
return fmt.Errorf("alarm delivery %s: %w", channel, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (d *Dispatcher) notify() {
|
|
select {
|
|
case d.wake <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (d *Dispatcher) trimTerminalLocked() {
|
|
if len(d.jobs) < d.maxJobs {
|
|
return
|
|
}
|
|
terminal := make([]deliveryJob, 0)
|
|
for _, job := range d.jobs {
|
|
if terminalDeliveryState(job.State) {
|
|
terminal = append(terminal, job)
|
|
}
|
|
}
|
|
sort.Slice(terminal, func(left, right int) bool {
|
|
if terminal[left].UpdatedAt.Equal(terminal[right].UpdatedAt) {
|
|
return terminal[left].DeliveryID < terminal[right].DeliveryID
|
|
}
|
|
return terminal[left].UpdatedAt.Before(terminal[right].UpdatedAt)
|
|
})
|
|
for _, job := range terminal {
|
|
if len(d.jobs) < d.maxJobs {
|
|
return
|
|
}
|
|
delete(d.jobs, job.DeliveryID)
|
|
}
|
|
}
|
|
|
|
func (d *Dispatcher) load() error {
|
|
pathInfo, err := os.Lstat(d.statePath)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return d.persistLocked()
|
|
}
|
|
return fmt.Errorf("inspect alarm delivery state: %w", err)
|
|
}
|
|
if pathInfo.Mode()&os.ModeSymlink != 0 || !pathInfo.Mode().IsRegular() {
|
|
return errors.New("alarm delivery state must be a bounded regular file")
|
|
}
|
|
file, err := os.Open(d.statePath)
|
|
if err != nil {
|
|
return fmt.Errorf("read alarm delivery state: %w", err)
|
|
}
|
|
info, err := file.Stat()
|
|
if err != nil {
|
|
_ = file.Close()
|
|
return fmt.Errorf("inspect alarm delivery state: %w", err)
|
|
}
|
|
if !info.Mode().IsRegular() || info.Size() < 1 || info.Size() > maxStateBytes {
|
|
_ = file.Close()
|
|
return errors.New("alarm delivery state must be a bounded regular file")
|
|
}
|
|
content, err := io.ReadAll(io.LimitReader(file, maxStateBytes+1))
|
|
if err != nil {
|
|
_ = file.Close()
|
|
return fmt.Errorf("read alarm delivery state: %w", err)
|
|
}
|
|
if err := file.Close(); err != nil {
|
|
return fmt.Errorf("close alarm delivery state: %w", err)
|
|
}
|
|
if len(content) > maxStateBytes {
|
|
return errors.New("alarm delivery state exceeds the configured limit")
|
|
}
|
|
var snapshot dispatcherSnapshot
|
|
decoder := json.NewDecoder(strings.NewReader(string(content)))
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(&snapshot); err != nil {
|
|
return fmt.Errorf("decode alarm delivery state: %w", err)
|
|
}
|
|
var extra any
|
|
if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) {
|
|
return errors.New("alarm delivery state must contain one JSON value")
|
|
}
|
|
if snapshot.FormatVersion != 1 || len(snapshot.Jobs) > d.maxJobs {
|
|
return errors.New("alarm delivery state is invalid or exceeds the configured limit")
|
|
}
|
|
now := d.clock().UTC()
|
|
for deliveryID, job := range snapshot.Jobs {
|
|
if deliveryID != job.DeliveryID || validateDeliveryID(deliveryID) != nil {
|
|
return errors.New("alarm delivery state contains an invalid delivery id")
|
|
}
|
|
normalized, err := NormalizeRequest(job.Request)
|
|
if err != nil || job.AcceptedAt.IsZero() || job.UpdatedAt.IsZero() || job.Attempts < 0 || job.Attempts > normalized.RetryCount+1 {
|
|
return fmt.Errorf("alarm delivery state job %s is invalid", deliveryID)
|
|
}
|
|
job.Request = normalized
|
|
switch job.State {
|
|
case "queued", "retry_pending", "delivered", "failed":
|
|
case "running":
|
|
job.State = "queued"
|
|
job.NextAttempt = now
|
|
default:
|
|
return fmt.Errorf("alarm delivery state job %s has invalid state", deliveryID)
|
|
}
|
|
d.jobs[deliveryID] = job
|
|
if sequence := idSequence(deliveryID); sequence > d.sequence {
|
|
d.sequence = sequence
|
|
}
|
|
}
|
|
if snapshot.Sequence > d.sequence {
|
|
d.sequence = snapshot.Sequence
|
|
}
|
|
return d.persistLocked()
|
|
}
|
|
|
|
func (d *Dispatcher) persistLocked() error {
|
|
snapshot := dispatcherSnapshot{FormatVersion: 1, Sequence: d.sequence, Jobs: make(map[string]deliveryJob, len(d.jobs))}
|
|
for deliveryID, job := range d.jobs {
|
|
snapshot.Jobs[deliveryID] = job
|
|
}
|
|
content, err := json.MarshalIndent(snapshot, "", " ")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := atomicfile.Write(d.statePath, append(content, '\n'), 0o600); err != nil {
|
|
return fmt.Errorf("persist alarm delivery state: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func ensureStateDirectory(directory string) error {
|
|
if info, err := os.Lstat(directory); err == nil {
|
|
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
|
|
return errors.New("alarm delivery state directory must be a real directory")
|
|
}
|
|
} else if os.IsNotExist(err) {
|
|
if err := os.MkdirAll(directory, 0o700); err != nil {
|
|
return err
|
|
}
|
|
} else {
|
|
return err
|
|
}
|
|
return os.Chmod(directory, 0o700)
|
|
}
|
|
|
|
func validateDeliveryID(value string) error {
|
|
if len(value) != len("alarm_delivery_20060102_000000") || !strings.HasPrefix(value, "alarm_delivery_") {
|
|
return errors.New("alarm delivery id is invalid")
|
|
}
|
|
dateStart := len("alarm_delivery_")
|
|
if _, err := time.Parse("20060102", value[dateStart:dateStart+8]); err != nil || value[dateStart+8] != '_' {
|
|
return errors.New("alarm delivery id is invalid")
|
|
}
|
|
for _, char := range value[len(value)-6:] {
|
|
if char < '0' || char > '9' {
|
|
return errors.New("alarm delivery id is invalid")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func terminalDeliveryState(state string) bool { return state == "delivered" || state == "failed" }
|
|
|
|
func resultFromJob(job deliveryJob) Result {
|
|
return Result{
|
|
DeliveryID: job.DeliveryID, State: job.State, Attempts: job.Attempts,
|
|
LastError: job.LastError, AcceptedAt: job.AcceptedAt, UpdatedAt: job.UpdatedAt, CompletedAt: job.CompletedAt,
|
|
}
|
|
}
|
|
|
|
func boundedDeliveryError(err error) string {
|
|
message := strings.TrimSpace(err.Error())
|
|
message = strings.ReplaceAll(message, "\r", " ")
|
|
message = strings.ReplaceAll(message, "\n", " ")
|
|
runes := []rune(message)
|
|
if len(runes) > 256 {
|
|
message = string(runes[:256])
|
|
}
|
|
if message == "" {
|
|
return "unknown delivery failure"
|
|
}
|
|
return message
|
|
}
|
|
|
|
func timePointer(value time.Time) *time.Time { return &value }
|
|
|
|
func idSequence(value string) int {
|
|
var sequence int
|
|
_, _ = fmt.Sscanf(value[len(value)-6:], "%d", &sequence)
|
|
return sequence
|
|
}
|
|
|