From 2bc2193a5939e4594c317dc92d3651540134bf91 Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 11:30:07 +0200 Subject: [PATCH 1/9] store: harden journal lifecycle and retention --- internal/api/server.go | 21 ++- internal/api/server_test.go | 38 +++++ internal/compactor/compactor.go | 44 ++++- internal/compactor/compactor_test.go | 49 ++++++ internal/store/lock_other.go | 16 ++ internal/store/lock_unix.go | 38 +++++ internal/store/store.go | 246 ++++++++++++++++++++++++--- internal/store/store_test.go | 180 ++++++++++++++++++++ 8 files changed, 607 insertions(+), 25 deletions(-) create mode 100644 internal/store/lock_other.go create mode 100644 internal/store/lock_unix.go diff --git a/internal/api/server.go b/internal/api/server.go index 379a259..2afed6b 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -52,7 +52,7 @@ func New(journal *store.Store, metrics metricsSource, logger *slog.Logger) *Serv } mux := http.NewServeMux() mux.HandleFunc("GET /healthz", server.health) - mux.HandleFunc("GET /readyz", server.health) + mux.HandleFunc("GET /readyz", server.ready) mux.HandleFunc("GET /metrics", server.renderMetrics) mux.HandleFunc("POST /v1/deliveries", server.create) mux.HandleFunc("GET /v1/deliveries", server.list) @@ -92,6 +92,10 @@ func (s *Server) create(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusInternalServerError, "persistence_failed", "delivery could not be persisted") return } + if errors.Is(err, store.ErrJournalFull) { + writeError(w, http.StatusInsufficientStorage, "journal_full", "journal admission limit reached") + return + } writeError(w, http.StatusBadRequest, "invalid_delivery", err.Error()) return } @@ -162,6 +166,15 @@ func (s *Server) health(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } +func (s *Server) ready(w http.ResponseWriter, _ *http.Request) { + if err := s.store.Ready(); err != nil { + s.log.Warn("journal not ready", "error", err) + writeError(w, http.StatusServiceUnavailable, "not_ready", "journal persistence is unavailable") + return + } + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + func (s *Server) renderMetrics(w http.ResponseWriter, _ *http.Request) { counts := s.store.Counts() journal := s.store.Stats() @@ -196,6 +209,12 @@ func (s *Server) renderMetrics(w http.ResponseWriter, _ *http.Request) { fmt.Fprintln(w, "# HELP spoold_journal_records Current physical journal record count.") fmt.Fprintln(w, "# TYPE spoold_journal_records gauge") fmt.Fprintf(w, "spoold_journal_records %d\n", journal.JournalRecords) + fmt.Fprintln(w, "# HELP spoold_journal_max_bytes Configured journal admission limit in bytes; zero is unlimited.") + fmt.Fprintln(w, "# TYPE spoold_journal_max_bytes gauge") + fmt.Fprintf(w, "spoold_journal_max_bytes %d\n", journal.MaxJournalBytes) + fmt.Fprintln(w, "# HELP spoold_journal_pruned_deliveries_total Terminal deliveries removed by retention.") + fmt.Fprintln(w, "# TYPE spoold_journal_pruned_deliveries_total counter") + fmt.Fprintf(w, "spoold_journal_pruned_deliveries_total %d\n", journal.PrunedDeliveries) fmt.Fprintln(w, "# HELP spoold_journal_compactions_total Journal compaction attempts by result.") fmt.Fprintln(w, "# TYPE spoold_journal_compactions_total counter") fmt.Fprintf(w, "spoold_journal_compactions_total{result=\"succeeded\"} %d\n", journal.CompactionsSucceeded) diff --git a/internal/api/server_test.go b/internal/api/server_test.go index 3159467..de6c7cc 100644 --- a/internal/api/server_test.go +++ b/internal/api/server_test.go @@ -94,6 +94,42 @@ func TestRejectsUnknownJSONField(t *testing.T) { } } +func TestReadinessReflectsJournalAvailability(t *testing.T) { + server, journal := newTestServer(t) + if got := request(t, server, http.MethodGet, "/readyz", ""); got.Code != http.StatusOK { + t.Fatalf("ready status = %d, body = %s", got.Code, got.Body.String()) + } + if err := journal.Close(); err != nil { + t.Fatal(err) + } + if got := request(t, server, http.MethodGet, "/healthz", ""); got.Code != http.StatusOK { + t.Fatalf("health status = %d, body = %s", got.Code, got.Body.String()) + } + if got := request(t, server, http.MethodGet, "/readyz", ""); got.Code != http.StatusServiceUnavailable { + t.Fatalf("unready status = %d, body = %s", got.Code, got.Body.String()) + } +} + +func TestCreateReturnsInsufficientStorageAtAdmissionLimit(t *testing.T) { + journal, err := store.OpenWithOptions(filepath.Join(t.TempDir(), "journal"), store.Options{ + MaxJournalBytes: 1, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { journal.Close() }) + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + server := New(journal, nil, logger) + + got := request(t, server, http.MethodPost, "/v1/deliveries", `{"targetUrl":"https://example.com"}`) + if got.Code != http.StatusInsufficientStorage { + t.Fatalf("status = %d, body = %s", got.Code, got.Body.String()) + } + if !strings.Contains(got.Body.String(), "journal_full") { + t.Fatalf("body = %s", got.Body.String()) + } +} + func TestStatusRecorderKeepsFirstResponseStatus(t *testing.T) { response := httptest.NewRecorder() recorder := &statusRecorder{ @@ -150,6 +186,8 @@ func TestMetricsExposeDeliveryAndJournalHealth(t *testing.T) { `spoold_deliveries{status="in_flight"} 1`, "spoold_journal_size_bytes " + strconv.FormatInt(stats.JournalSizeBytes, 10), "spoold_journal_records 1", + "spoold_journal_max_bytes 0", + "spoold_journal_pruned_deliveries_total 0", `spoold_journal_compactions_total{result="succeeded"} 1`, `spoold_journal_compactions_total{result="failed"} 0`, } { diff --git a/internal/compactor/compactor.go b/internal/compactor/compactor.go index c34649e..cbe60c6 100644 --- a/internal/compactor/compactor.go +++ b/internal/compactor/compactor.go @@ -12,16 +12,19 @@ import ( const ( DefaultThresholdBytes int64 = 64 << 20 DefaultCheckInterval = time.Minute + DefaultRetention = 7 * 24 * time.Hour ) type Config struct { ThresholdBytes int64 CheckInterval time.Duration + Retention time.Duration } type journalStore interface { Stats() store.Stats Compact() error + PruneTerminal(time.Time) (int, error) } type Compactor struct { @@ -29,6 +32,7 @@ type Compactor struct { log *slog.Logger threshold int64 interval time.Duration + retention time.Duration wg sync.WaitGroup } @@ -41,11 +45,12 @@ func New(journal journalStore, logger *slog.Logger, config Config) *Compactor { log: logger, threshold: config.ThresholdBytes, interval: config.CheckInterval, + retention: config.Retention, } } func (c *Compactor) Start(ctx context.Context) { - if c.threshold <= 0 { + if c.threshold <= 0 && c.retention <= 0 { return } c.wg.Add(1) @@ -66,12 +71,37 @@ func (c *Compactor) run(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - c.compactIfNeeded() + c.maintain() + } + } +} + +func (c *Compactor) maintain() { + if c.retention > 0 { + started := time.Now() + pruned, err := c.store.PruneTerminal(started.Add(-c.retention)) + if err != nil { + c.log.Warn("journal retention failed", "error", err) + return + } + if pruned > 0 { + if err := c.compact("retention", started); err != nil { + return + } + c.log.Info("terminal deliveries pruned", + "deliveries", pruned, + "retention", c.retention, + ) + return } } + c.compactIfNeeded() } func (c *Compactor) compactIfNeeded() { + if c.threshold <= 0 { + return + } before := c.store.Stats() if before.JournalSizeBytes < c.threshold { return @@ -81,18 +111,25 @@ func (c *Compactor) compactIfNeeded() { } started := time.Now() + _ = c.compact("redundancy", started) +} + +func (c *Compactor) compact(reason string, started time.Time) error { + before := c.store.Stats() if err := c.store.Compact(); err != nil { c.log.Warn("journal compaction failed", "error", err, + "reason", reason, "duration_ms", time.Since(started).Milliseconds(), "size_bytes", before.JournalSizeBytes, "records", before.JournalRecords, ) - return + return err } after := c.store.Stats() c.log.Info("journal compacted", + "reason", reason, "duration_ms", time.Since(started).Milliseconds(), "size_bytes_before", before.JournalSizeBytes, "size_bytes_after", after.JournalSizeBytes, @@ -101,6 +138,7 @@ func (c *Compactor) compactIfNeeded() { "records_after", after.JournalRecords, "records_reduced", reducedUint64(before.JournalRecords, after.JournalRecords), ) + return nil } func reducedInt64(before, after int64) int64 { diff --git a/internal/compactor/compactor_test.go b/internal/compactor/compactor_test.go index 4d09b80..f6f5ffa 100644 --- a/internal/compactor/compactor_test.go +++ b/internal/compactor/compactor_test.go @@ -147,11 +147,40 @@ func TestCompactionFailureIsNonFatal(t *testing.T) { compactor.Wait() } +func TestRetentionPrunesTerminalDeliveriesAndCompactsTombstones(t *testing.T) { + now := time.Now() + journal := &fakeStore{ + stats: store.Stats{ + JournalSizeBytes: 100, + JournalRecords: 2, + LiveDeliveries: 2, + }, + items: []time.Time{ + now.Add(-2 * time.Hour), + now.Add(-30 * time.Minute), + }, + } + compactor := New(journal, discardLogger(), Config{ + ThresholdBytes: 0, + Retention: time.Hour, + }) + compactor.maintain() + + if got := journal.compactCalls(); got != 1 { + t.Fatalf("Compact() calls = %d, want 1", got) + } + stats := journal.Stats() + if stats.LiveDeliveries != 1 || stats.JournalRecords != 1 || stats.PrunedDeliveries != 1 { + t.Fatalf("stats = %#v", stats) + } +} + type fakeStore struct { mu sync.Mutex stats store.Stats calls int failures int + items []time.Time } func (s *fakeStore) Stats() store.Stats { @@ -181,6 +210,26 @@ func (s *fakeStore) compactCalls() int { return s.calls } +func (s *fakeStore) PruneTerminal(before time.Time) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + kept := s.items[:0] + pruned := 0 + for _, updatedAt := range s.items { + if updatedAt.After(before) { + kept = append(kept, updatedAt) + } else { + pruned++ + } + } + s.items = kept + s.stats.LiveDeliveries -= uint64(pruned) + s.stats.JournalRecords += uint64(pruned) + s.stats.PrunedDeliveries += uint64(pruned) + return pruned, nil +} + func waitForCompactions(t *testing.T, store *fakeStore, count int) { t.Helper() deadline := time.Now().Add(time.Second) diff --git a/internal/store/lock_other.go b/internal/store/lock_other.go new file mode 100644 index 0000000..79dc586 --- /dev/null +++ b/internal/store/lock_other.go @@ -0,0 +1,16 @@ +//go:build !unix + +package store + +import ( + "errors" + "os" +) + +func acquireJournalLock(string) (*os.File, error) { + return nil, errors.New("journal locking is unsupported on this platform") +} + +func releaseJournalLock(*os.File) error { + return nil +} diff --git a/internal/store/lock_unix.go b/internal/store/lock_unix.go new file mode 100644 index 0000000..24eb1bd --- /dev/null +++ b/internal/store/lock_unix.go @@ -0,0 +1,38 @@ +//go:build unix + +package store + +import ( + "errors" + "fmt" + "os" + "syscall" +) + +func acquireJournalLock(path string) (*os.File, error) { + lock, err := os.OpenFile(path+".lock", os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, fmt.Errorf("open journal lock: %w", err) + } + if err := lock.Chmod(0o600); err != nil { + lock.Close() + return nil, fmt.Errorf("set journal lock permissions: %w", err) + } + if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + lock.Close() + if errors.Is(err, syscall.EWOULDBLOCK) || errors.Is(err, syscall.EAGAIN) { + return nil, ErrJournalLocked + } + return nil, fmt.Errorf("lock journal: %w", err) + } + return lock, nil +} + +func releaseJournalLock(lock *os.File) error { + if lock == nil { + return nil + } + unlockErr := syscall.Flock(int(lock.Fd()), syscall.LOCK_UN) + closeErr := lock.Close() + return errors.Join(unlockErr, closeErr) +} diff --git a/internal/store/store.go b/internal/store/store.go index cb05ba5..de1d8a6 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -2,6 +2,7 @@ package store import ( "bufio" + "bytes" "encoding/json" "errors" "fmt" @@ -22,25 +23,38 @@ var ( ErrInvalidTransition = errors.New("invalid delivery state transition") ErrStaleLease = errors.New("delivery lease is no longer current") ErrPersistence = errors.New("delivery persistence failed") + ErrJournalFull = errors.New("journal admission limit reached") + ErrJournalLocked = errors.New("journal is already owned by another spoold process") ) +const currentJournalVersion = 2 + +type Options struct { + MaxJournalBytes int64 +} + type Store struct { mu sync.Mutex path string file *os.File + lock *os.File items map[string]delivery.Delivery hashes map[string]string keyToID map[string]string + maxJournalBytes int64 + persistenceErr error records uint64 + prunedDeliveries uint64 compactionsSucceeded uint64 compactionsFailed uint64 compactionHook func(compactionStage) error } type journalRecord struct { - Version int `json:"version"` - Delivery delivery.Delivery `json:"delivery"` - RequestHash string `json:"requestHash,omitempty"` + Version int `json:"version"` + Delivery *delivery.Delivery `json:"delivery,omitempty"` + RequestHash string `json:"requestHash,omitempty"` + DeletedID string `json:"deletedId,omitempty"` } type Counts map[delivery.Status]int @@ -49,6 +63,8 @@ type Stats struct { JournalSizeBytes int64 JournalRecords uint64 LiveDeliveries uint64 + MaxJournalBytes int64 + PrunedDeliveries uint64 CompactionsSucceeded uint64 CompactionsFailed uint64 } @@ -62,37 +78,80 @@ const ( ) func Open(path string) (*Store, error) { + return OpenWithOptions(path, Options{}) +} + +func OpenWithOptions(path string, options Options) (*Store, error) { + if options.MaxJournalBytes < 0 { + return nil, errors.New("maximum journal size must not be negative") + } dir := filepath.Dir(path) if err := os.MkdirAll(dir, 0o750); err != nil { return nil, fmt.Errorf("create journal directory: %w", err) } + + lock, err := acquireJournalLock(path) + if err != nil { + return nil, err + } + releaseLock := true + defer func() { + if releaseLock { + _ = releaseJournalLock(lock) + } + }() + if err := removeAbandonedCompactions(path); err != nil { return nil, err } + _, statErr := os.Stat(path) + newJournal := errors.Is(statErr, os.ErrNotExist) + if statErr != nil && !newJournal { + return nil, fmt.Errorf("stat journal: %w", statErr) + } + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600) if err != nil { return nil, fmt.Errorf("open journal: %w", err) } + if newJournal { + if err := syncDirectory(dir); err != nil { + file.Close() + _ = os.Remove(path) + return nil, fmt.Errorf("persist new journal: %w", err) + } + } store := &Store{ - path: path, - file: file, - items: make(map[string]delivery.Delivery), - hashes: make(map[string]string), - keyToID: make(map[string]string), + path: path, + file: file, + lock: lock, + items: make(map[string]delivery.Delivery), + hashes: make(map[string]string), + keyToID: make(map[string]string), + maxJournalBytes: options.MaxJournalBytes, } if err := store.replay(); err != nil { file.Close() return nil, err } + releaseLock = false return store, nil } func (s *Store) Close() error { s.mu.Lock() defer s.mu.Unlock() - return s.file.Close() + + var fileErr error + if s.file != nil { + fileErr = s.file.Close() + s.file = nil + } + lockErr := releaseJournalLock(s.lock) + s.lock = nil + return errors.Join(fileErr, lockErr) } func (s *Store) Create(req delivery.CreateRequest, now time.Time) (delivery.Delivery, bool, error) { @@ -113,8 +172,11 @@ func (s *Store) Create(req delivery.CreateRequest, now time.Time) (delivery.Deli } } + if err := s.checkAdmissionLocked(candidate, requestHash); err != nil { + return delivery.Delivery{}, false, err + } if err := s.appendLocked(candidate, requestHash); err != nil { - return delivery.Delivery{}, false, fmt.Errorf("%w: %v", ErrPersistence, err) + return delivery.Delivery{}, false, err } s.setLocked(candidate, requestHash) return delivery.Clone(candidate), true, nil @@ -294,18 +356,72 @@ func (s *Store) Stats() Stats { defer s.mu.Unlock() var size int64 - if info, err := s.file.Stat(); err == nil { - size = info.Size() + if s.file != nil { + if info, err := s.file.Stat(); err == nil { + size = info.Size() + } } return Stats{ JournalSizeBytes: size, JournalRecords: s.records, LiveDeliveries: uint64(len(s.items)), + MaxJournalBytes: s.maxJournalBytes, + PrunedDeliveries: s.prunedDeliveries, CompactionsSucceeded: s.compactionsSucceeded, CompactionsFailed: s.compactionsFailed, } } +func (s *Store) Ready() error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.persistenceErr != nil { + return s.persistenceErr + } + if s.file == nil { + return errors.New("journal is closed") + } + if _, err := s.file.Stat(); err != nil { + return fmt.Errorf("stat journal: %w", err) + } + return nil +} + +func (s *Store) PruneTerminal(before time.Time) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + ids := make([]string, 0) + for id, item := range s.items { + terminal := item.Status == delivery.StatusSucceeded || + item.Status == delivery.StatusFailed || + item.Status == delivery.StatusCanceled + if terminal && !item.UpdatedAt.After(before) { + ids = append(ids, id) + } + } + if len(ids) == 0 { + return 0, nil + } + sort.Strings(ids) + + var records bytes.Buffer + for _, id := range ids { + if err := writeDeleteRecord(&records, id); err != nil { + return 0, err + } + } + if err := s.appendDataLocked(records.Bytes(), uint64(len(ids))); err != nil { + return 0, err + } + for _, id := range ids { + s.deleteLocked(id) + } + s.prunedDeliveries += uint64(len(ids)) + return len(ids), nil +} + func (s *Store) Compact() (err error) { s.mu.Lock() defer s.mu.Unlock() @@ -315,6 +431,9 @@ func (s *Store) Compact() (err error) { s.compactionsFailed++ } }() + if s.file == nil { + return errors.New("compact closed journal") + } temp, err := os.CreateTemp(filepath.Dir(s.path), compactionPrefix(s.path)) if err != nil { @@ -380,6 +499,7 @@ func (s *Store) Compact() (err error) { return err } adopted = true + s.persistenceErr = nil s.compactionsSucceeded++ return nil } @@ -411,25 +531,74 @@ func (s *Store) setLocked(item delivery.Delivery, requestHash string) { } } +func (s *Store) deleteLocked(id string) { + item, ok := s.items[id] + if !ok { + return + } + delete(s.items, id) + delete(s.hashes, id) + if item.IdempotencyKey != "" && s.keyToID[item.IdempotencyKey] == id { + delete(s.keyToID, item.IdempotencyKey) + } +} + func (s *Store) appendLocked(item delivery.Delivery, requestHash string) error { - if err := writeRecord(s.file, item, requestHash); err != nil { + data, err := marshalRecord(item, requestHash) + if err != nil { return err } + return s.appendDataLocked(data, 1) +} + +func (s *Store) appendDataLocked(data []byte, records uint64) error { + if s.persistenceErr != nil { + return s.persistenceErr + } + if s.file == nil { + return s.failPersistenceLocked(errors.New("journal is closed")) + } + if _, err := s.file.Write(data); err != nil { + return s.failPersistenceLocked(fmt.Errorf("append journal record: %w", err)) + } if err := s.file.Sync(); err != nil { - return fmt.Errorf("sync journal: %w", err) + return s.failPersistenceLocked(fmt.Errorf("sync journal: %w", err)) } - s.records++ + s.records += records return nil } func writeRecord(writer io.Writer, item delivery.Delivery, requestHash string) error { + data, err := marshalRecord(item, requestHash) + if err != nil { + return err + } + if _, err := writer.Write(data); err != nil { + return fmt.Errorf("append journal record: %w", err) + } + return nil +} + +func marshalRecord(item delivery.Delivery, requestHash string) ([]byte, error) { data, err := json.Marshal(journalRecord{ - Version: 1, - Delivery: item, + Version: currentJournalVersion, + Delivery: &item, RequestHash: requestHash, }) if err != nil { - return fmt.Errorf("encode journal record: %w", err) + return nil, fmt.Errorf("encode journal record: %w", err) + } + data = append(data, '\n') + return data, nil +} + +func writeDeleteRecord(writer io.Writer, id string) error { + data, err := json.Marshal(journalRecord{ + Version: currentJournalVersion, + DeletedID: id, + }) + if err != nil { + return fmt.Errorf("encode journal deletion: %w", err) } data = append(data, '\n') if _, err := writer.Write(data); err != nil { @@ -482,16 +651,51 @@ func (s *Store) replay() error { } func (s *Store) applyRecord(record journalRecord, line int) error { - if record.Version != 1 { + if record.Version != 1 && record.Version != currentJournalVersion { return fmt.Errorf("journal line %d uses unsupported version %d", line, record.Version) } - if record.Delivery.ID == "" { + if record.DeletedID != "" { + if record.Version < 2 { + return fmt.Errorf("journal line %d uses a deletion with version %d", line, record.Version) + } + if record.Delivery != nil { + return fmt.Errorf("journal line %d contains both a delivery and deletion", line) + } + s.deleteLocked(record.DeletedID) + return nil + } + if record.Delivery == nil || record.Delivery.ID == "" { return fmt.Errorf("journal line %d has no delivery id", line) } - s.setLocked(record.Delivery, record.RequestHash) + s.setLocked(*record.Delivery, record.RequestHash) + return nil +} + +func (s *Store) checkAdmissionLocked(item delivery.Delivery, requestHash string) error { + if s.maxJournalBytes == 0 { + return nil + } + data, err := marshalRecord(item, requestHash) + if err != nil { + return err + } + info, err := s.file.Stat() + if err != nil { + return s.failPersistenceLocked(fmt.Errorf("stat journal: %w", err)) + } + if info.Size()+int64(len(data)) > s.maxJournalBytes { + return ErrJournalFull + } return nil } +func (s *Store) failPersistenceLocked(err error) error { + if s.persistenceErr == nil { + s.persistenceErr = fmt.Errorf("%w: %w", ErrPersistence, err) + } + return s.persistenceErr +} + func (s *Store) adoptCompactedLocked(file *os.File, records uint64) error { old := s.file s.file = file diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 02e03bc..1470bf3 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -79,6 +79,186 @@ func TestIdempotencyKeyRejectsDifferentRequest(t *testing.T) { } } +func TestOpenRejectsConcurrentJournalOwner(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + first, err := Open(path) + if err != nil { + t.Fatal(err) + } + if _, err := Open(path); !errors.Is(err, ErrJournalLocked) { + t.Fatalf("second Open() error = %v, want %v", err, ErrJournalLocked) + } + if err := first.Close(); err != nil { + t.Fatal(err) + } + + reopened, err := Open(path) + if err != nil { + t.Fatalf("Open() after owner closed: %v", err) + } + if err := reopened.Close(); err != nil { + t.Fatal(err) + } +} + +func TestJournalAdmissionLimitRejectsOnlyNewDeliveries(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + request := delivery.CreateRequest{ + IdempotencyKey: "first", + TargetURL: "https://example.com/first", + } + now := time.Date(2026, 7, 18, 10, 0, 0, 0, time.UTC) + + initial, err := Open(path) + if err != nil { + t.Fatal(err) + } + first, _, err := initial.Create(request, now) + if err != nil { + t.Fatal(err) + } + if err := initial.Close(); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + + bounded, err := OpenWithOptions(path, Options{MaxJournalBytes: info.Size()}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { bounded.Close() }) + duplicate, created, err := bounded.Create(request, now.Add(time.Minute)) + if err != nil || created || duplicate.ID != first.ID { + t.Fatalf("idempotent Create() = (%q, %v, %v)", duplicate.ID, created, err) + } + if _, _, err := bounded.Create(delivery.CreateRequest{ + TargetURL: "https://example.com/second", + }, now); !errors.Is(err, ErrJournalFull) { + t.Fatalf("new Create() error = %v, want %v", err, ErrJournalFull) + } + if claimed, err := bounded.ClaimDue(now, time.Minute, 1); err != nil || len(claimed) != 1 { + t.Fatalf("ClaimDue() = %#v, %v", claimed, err) + } + if got := bounded.Stats().MaxJournalBytes; got != info.Size() { + t.Fatalf("maximum journal bytes = %d, want %d", got, info.Size()) + } +} + +func TestPersistenceFailureMarksStoreUnreadyUntilCompactionRepairsIt(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + store, err := Open(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { store.Close() }) + + original := store.file + readOnly, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + store.file = readOnly + if _, _, err := store.Create(delivery.CreateRequest{ + TargetURL: "https://example.com/fails", + }, time.Now()); !errors.Is(err, ErrPersistence) { + t.Fatalf("Create() error = %v, want %v", err, ErrPersistence) + } + if err := store.Ready(); !errors.Is(err, ErrPersistence) { + t.Fatalf("Ready() error = %v, want %v", err, ErrPersistence) + } + + if err := store.Compact(); err != nil { + t.Fatalf("repairing Compact(): %v", err) + } + if err := original.Close(); err != nil { + t.Fatal(err) + } + if err := store.Ready(); err != nil { + t.Fatalf("Ready() after compaction: %v", err) + } + if _, created, err := store.Create(delivery.CreateRequest{ + TargetURL: "https://example.com/works", + }, time.Now()); err != nil || !created { + t.Fatalf("Create() after repair = (%v, %v)", created, err) + } +} + +func TestPruneTerminalRemovesStateAndIdempotencyKeysAcrossReplay(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + store, err := Open(path) + if err != nil { + t.Fatal(err) + } + now := time.Date(2026, 7, 18, 10, 0, 0, 0, time.UTC) + + succeeded := createTestDelivery(t, store, "succeeded", now) + claimed, err := store.ClaimDue(now, time.Minute, 1) + if err != nil || len(claimed) != 1 { + t.Fatalf("claim succeeded = %#v, %v", claimed, err) + } + if err := store.Succeed(succeeded.ID, claimed[0].Attempts, 204, now.Add(time.Second)); err != nil { + t.Fatal(err) + } + + failed := createTestDelivery(t, store, "failed", now.Add(time.Minute)) + claimed, err = store.ClaimDue(now.Add(time.Minute), time.Minute, 1) + if err != nil || len(claimed) != 1 { + t.Fatalf("claim failed = %#v, %v", claimed, err) + } + if err := store.Fail(failed.ID, claimed[0].Attempts, 503, "failed", time.Time{}, true, now.Add(time.Minute+time.Second)); err != nil { + t.Fatal(err) + } + + canceled := createTestDelivery(t, store, "canceled", now.Add(2*time.Minute)) + if _, err := store.Cancel(canceled.ID, now.Add(2*time.Minute+time.Second)); err != nil { + t.Fatal(err) + } + pending := createTestDelivery(t, store, "pending", now.Add(3*time.Minute)) + + pruned, err := store.PruneTerminal(now.Add(10 * time.Minute)) + if err != nil { + t.Fatal(err) + } + if pruned != 3 { + t.Fatalf("pruned deliveries = %d, want 3", pruned) + } + for _, id := range []string{succeeded.ID, failed.ID, canceled.ID} { + if _, err := store.Get(id); !errors.Is(err, ErrNotFound) { + t.Fatalf("Get(%q) error = %v, want %v", id, err, ErrNotFound) + } + } + if _, err := store.Get(pending.ID); err != nil { + t.Fatalf("pending delivery was pruned: %v", err) + } + + replacement, created, err := store.Create(delivery.CreateRequest{ + IdempotencyKey: "succeeded", + TargetURL: "https://example.com/succeeded", + }, now.Add(time.Hour)) + if err != nil || !created || replacement.ID == succeeded.ID { + t.Fatalf("replacement Create() = (%q, %v, %v)", replacement.ID, created, err) + } + if got := store.Stats().PrunedDeliveries; got != 3 { + t.Fatalf("pruned delivery metric = %d, want 3", got) + } + + before := store.List("") + if err := store.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { reopened.Close() }) + if got := reopened.List(""); !reflect.DeepEqual(got, before) { + t.Fatalf("replayed state differs:\ngot %#v\nwant %#v", got, before) + } +} + func TestExpiredLeaseCanBeReclaimedAndRejectsStaleCompletion(t *testing.T) { store := openTestStore(t) now := time.Date(2026, 7, 16, 10, 0, 0, 0, time.UTC) From a77930077a1b69e46b00a8e17e405e66810af87d Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 11:43:00 +0200 Subject: [PATCH 2/9] runtime: complete the local delivery contract --- cmd/spoold/main.go | 219 ++++++++++++++++++++++++----- cmd/spoold/main_test.go | 78 ++++++++++ internal/api/server.go | 29 +++- internal/api/server_test.go | 59 +++++++- internal/delivery/delivery.go | 86 +++++++++-- internal/delivery/delivery_test.go | 49 ++++++- internal/spoolctl/spoolctl.go | 94 ++++++++++--- internal/spoolctl/spoolctl_test.go | 78 +++++++++- internal/store/store.go | 28 ++-- internal/store/store_test.go | 38 ++++- internal/worker/worker.go | 59 +++++++- internal/worker/worker_test.go | 88 ++++++++++++ 12 files changed, 823 insertions(+), 82 deletions(-) create mode 100644 cmd/spoold/main_test.go diff --git a/cmd/spoold/main.go b/cmd/spoold/main.go index 0efc4f2..45ef855 100644 --- a/cmd/spoold/main.go +++ b/cmd/spoold/main.go @@ -4,10 +4,14 @@ import ( "context" "errors" "flag" + "fmt" + "io" "log/slog" + "net" "net/http" "os" "os/signal" + "path/filepath" "syscall" "time" @@ -18,45 +22,71 @@ import ( "github.com/Rionlyu/spoold/internal/worker" ) +const defaultMaxJournalBytes int64 = 1 << 30 + +type config struct { + listen string + unixSocket string + journalPath string + maxJournalBytes int64 + terminalRetention time.Duration + concurrency int + perTargetWorkers int + allowPrivateTargets bool + requestTimeout time.Duration + shutdownTimeout time.Duration + compactThreshold int64 + compactInterval time.Duration +} + func main() { - var ( - listen = flag.String("listen", "127.0.0.1:8080", "HTTP listen address") - journalPath = flag.String("journal", "data/spoold.journal", "append-only journal path") - concurrency = flag.Int("workers", 4, "number of delivery workers") - allowPrivateTargets = flag.Bool("allow-private-targets", false, "allow private and loopback delivery targets") - requestTimeout = flag.Duration("request-timeout", 10*time.Second, "outbound request timeout") - shutdownTimeout = flag.Duration("shutdown-timeout", 10*time.Second, "graceful shutdown timeout") - compactThreshold = flag.Int64("compact-threshold-bytes", compactor.DefaultThresholdBytes, "minimum journal size for compaction (0 disables)") - compactInterval = flag.Duration("compact-check-interval", compactor.DefaultCheckInterval, "journal compaction check interval") - ) - flag.Parse() - - logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) - slog.SetDefault(logger) - - journal, err := store.Open(*journalPath) + os.Exit(run(os.Args[1:], os.Stdout, os.Stderr)) +} + +func run(args []string, stdout, stderr io.Writer) int { + cfg, code := parseConfig(args, stderr) + if code != 0 { + return code + } + + logger := slog.New(slog.NewJSONHandler(stdout, nil)) + journal, err := store.Open(cfg.journalPath, store.Options{ + MaxJournalBytes: cfg.maxJournalBytes, + }) if err != nil { logger.Error("open journal", "error", err) - os.Exit(1) + return 1 } - defer journal.Close() + + listener, cleanupListener, err := openListener(cfg.listen, cfg.unixSocket) + if err != nil { + logger.Error("listen", "error", err) + _ = journal.Close() + return 1 + } + defer func() { + if err := cleanupListener(); err != nil { + logger.Warn("clean up listener", "error", err) + } + }() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - client := target.NewClient(*allowPrivateTargets, *requestTimeout) + client := target.NewClient(cfg.allowPrivateTargets, cfg.requestTimeout) pool := worker.New(journal, client, logger, worker.Config{ - Concurrency: *concurrency, + Concurrency: cfg.concurrency, + PerTarget: cfg.perTargetWorkers, }) pool.Start(ctx) journalCompactor := compactor.New(journal, logger, compactor.Config{ - ThresholdBytes: *compactThreshold, - CheckInterval: *compactInterval, + ThresholdBytes: cfg.compactThreshold, + CheckInterval: cfg.compactInterval, + Retention: cfg.terminalRetention, }) journalCompactor.Start(ctx) httpServer := &http.Server{ - Addr: *listen, Handler: api.New(journal, pool, logger), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, @@ -66,33 +96,156 @@ func main() { serverErrors := make(chan error, 1) go func() { logger.Info("spoold started", - "listen", *listen, - "journal", *journalPath, - "workers", *concurrency, - "allow_private_targets", *allowPrivateTargets, - "compact_threshold_bytes", *compactThreshold, - "compact_check_interval", *compactInterval, + "network", listener.Addr().Network(), + "address", listener.Addr().String(), + "journal", cfg.journalPath, + "journal_max_bytes", cfg.maxJournalBytes, + "terminal_retention", cfg.terminalRetention, + "workers", cfg.concurrency, + "per_target_workers", cfg.perTargetWorkers, + "allow_private_targets", cfg.allowPrivateTargets, + "compact_threshold_bytes", cfg.compactThreshold, + "compact_check_interval", cfg.compactInterval, ) - serverErrors <- httpServer.ListenAndServe() + serverErrors <- httpServer.Serve(listener) }() + exitCode := 0 select { case <-ctx.Done(): logger.Info("shutdown requested") case err := <-serverErrors: if !errors.Is(err, http.ErrServerClosed) { logger.Error("HTTP server stopped", "error", err) - stop() + exitCode = 1 } + stop() } - shutdownCtx, cancel := context.WithTimeout(context.Background(), *shutdownTimeout) - defer cancel() + shutdownCtx, cancel := context.WithTimeout(context.Background(), cfg.shutdownTimeout) if err := httpServer.Shutdown(shutdownCtx); err != nil { logger.Error("graceful HTTP shutdown", "error", err) + exitCode = 1 } + cancel() stop() pool.Wait() journalCompactor.Wait() + if err := journal.Close(); err != nil { + logger.Error("close journal", "error", err) + exitCode = 1 + } logger.Info("spoold stopped") + return exitCode +} + +func parseConfig(args []string, stderr io.Writer) (config, int) { + var cfg config + flags := flag.NewFlagSet("spoold", flag.ContinueOnError) + flags.SetOutput(stderr) + flags.StringVar(&cfg.listen, "listen", "127.0.0.1:8080", "HTTP listen address") + flags.StringVar(&cfg.unixSocket, "unix-socket", "", "owner-only Unix socket path (overrides -listen)") + flags.StringVar(&cfg.journalPath, "journal", "data/spoold.journal", "append-only journal path") + flags.Int64Var(&cfg.maxJournalBytes, "max-journal-bytes", defaultMaxJournalBytes, "reject new deliveries when the journal reaches this size (0 disables)") + flags.DurationVar(&cfg.terminalRetention, "terminal-retention", compactor.DefaultRetention, "retain succeeded, failed, and canceled deliveries for this duration (0 retains forever)") + flags.IntVar(&cfg.concurrency, "workers", 4, "number of delivery workers") + flags.IntVar(&cfg.perTargetWorkers, "per-target-workers", 1, "maximum concurrent requests to one target origin") + flags.BoolVar(&cfg.allowPrivateTargets, "allow-private-targets", false, "allow private and loopback delivery targets") + flags.DurationVar(&cfg.requestTimeout, "request-timeout", 10*time.Second, "outbound request timeout") + flags.DurationVar(&cfg.shutdownTimeout, "shutdown-timeout", 10*time.Second, "graceful shutdown timeout") + flags.Int64Var(&cfg.compactThreshold, "compact-threshold-bytes", compactor.DefaultThresholdBytes, "minimum journal size for compaction (0 disables size-based compaction)") + flags.DurationVar(&cfg.compactInterval, "compact-check-interval", compactor.DefaultCheckInterval, "journal maintenance interval") + if err := flags.Parse(args); err != nil { + return config{}, 2 + } + if flags.NArg() != 0 { + fmt.Fprintln(stderr, "spoold: positional arguments are not supported") + return config{}, 2 + } + + switch { + case cfg.journalPath == "": + fmt.Fprintln(stderr, "spoold: -journal must not be empty") + case cfg.maxJournalBytes < 0: + fmt.Fprintln(stderr, "spoold: -max-journal-bytes must not be negative") + case cfg.terminalRetention < 0: + fmt.Fprintln(stderr, "spoold: -terminal-retention must not be negative") + case cfg.concurrency < 1: + fmt.Fprintln(stderr, "spoold: -workers must be at least 1") + case cfg.perTargetWorkers < 1 || cfg.perTargetWorkers > cfg.concurrency: + fmt.Fprintln(stderr, "spoold: -per-target-workers must be between 1 and -workers") + case cfg.requestTimeout <= 0: + fmt.Fprintln(stderr, "spoold: -request-timeout must be positive") + case cfg.shutdownTimeout <= 0: + fmt.Fprintln(stderr, "spoold: -shutdown-timeout must be positive") + case cfg.compactThreshold < 0: + fmt.Fprintln(stderr, "spoold: -compact-threshold-bytes must not be negative") + case cfg.compactInterval <= 0: + fmt.Fprintln(stderr, "spoold: -compact-check-interval must be positive") + case cfg.unixSocket == "" && cfg.listen == "": + fmt.Fprintln(stderr, "spoold: -listen must not be empty") + default: + return cfg, 0 + } + return config{}, 2 +} + +func openListener(address, socketPath string) (net.Listener, func() error, error) { + if socketPath == "" { + listener, err := net.Listen("tcp", address) + if err != nil { + return nil, nil, fmt.Errorf("listen on %q: %w", address, err) + } + return listener, func() error { + if err := listener.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + return err + } + return nil + }, nil + } + if !filepath.IsAbs(socketPath) { + return nil, nil, errors.New("unix socket path must be absolute") + } + if err := os.MkdirAll(filepath.Dir(socketPath), 0o700); err != nil { + return nil, nil, fmt.Errorf("create Unix socket directory: %w", err) + } + if info, err := os.Lstat(socketPath); err == nil { + if info.Mode()&os.ModeSocket == 0 { + return nil, nil, fmt.Errorf("unix socket path %q exists and is not a socket", socketPath) + } + connection, dialErr := net.DialTimeout("unix", socketPath, 200*time.Millisecond) + if dialErr == nil { + connection.Close() + return nil, nil, fmt.Errorf("unix socket %q is already in use", socketPath) + } + if !errors.Is(dialErr, syscall.ECONNREFUSED) { + return nil, nil, fmt.Errorf("inspect existing Unix socket %q: %w", socketPath, dialErr) + } + if err := os.Remove(socketPath); err != nil { + return nil, nil, fmt.Errorf("remove stale Unix socket: %w", err) + } + } else if !errors.Is(err, os.ErrNotExist) { + return nil, nil, fmt.Errorf("inspect Unix socket: %w", err) + } + + listener, err := net.Listen("unix", socketPath) + if err != nil { + return nil, nil, fmt.Errorf("listen on Unix socket %q: %w", socketPath, err) + } + if err := os.Chmod(socketPath, 0o600); err != nil { + listener.Close() + _ = os.Remove(socketPath) + return nil, nil, fmt.Errorf("set Unix socket permissions: %w", err) + } + return listener, func() error { + var closeErr error + if err := listener.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + closeErr = err + } + removeErr := os.Remove(socketPath) + if errors.Is(removeErr, os.ErrNotExist) { + removeErr = nil + } + return errors.Join(closeErr, removeErr) + }, nil } diff --git a/cmd/spoold/main_test.go b/cmd/spoold/main_test.go new file mode 100644 index 0000000..c956980 --- /dev/null +++ b/cmd/spoold/main_test.go @@ -0,0 +1,78 @@ +package main + +import ( + "bytes" + "net" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestRunReturnsFailureWhenAddressIsAlreadyInUse(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + + var stdout, stderr bytes.Buffer + code := run([]string{ + "-listen", listener.Addr().String(), + "-journal", filepath.Join(t.TempDir(), "journal"), + }, &stdout, &stderr) + if code != 1 { + t.Fatalf("exit code = %d, stdout = %s, stderr = %s", code, stdout.String(), stderr.String()) + } + if !strings.Contains(stdout.String(), `"msg":"listen"`) { + t.Fatalf("stdout = %s", stdout.String()) + } +} + +func TestParseConfigRejectsUnsafeValues(t *testing.T) { + tests := [][]string{ + {"-workers", "0"}, + {"-workers", "2", "-per-target-workers", "3"}, + {"-max-journal-bytes", "-1"}, + {"-terminal-retention", "-1s"}, + {"-request-timeout", "0"}, + {"unexpected"}, + } + for _, args := range tests { + t.Run(strings.Join(args, "_"), func(t *testing.T) { + var stderr bytes.Buffer + if _, code := parseConfig(args, &stderr); code != 2 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + }) + } +} + +func TestOpenListenerCreatesOwnerOnlyUnixSocket(t *testing.T) { + socketDir, err := os.MkdirTemp("/tmp", "spoold-listener-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(socketDir) }) + socketPath := filepath.Join(socketDir, "spoold.sock") + listener, cleanup, err := openListener("", socketPath) + if err != nil { + t.Fatal(err) + } + info, err := os.Stat(socketPath) + if err != nil { + t.Fatal(err) + } + if got := info.Mode().Perm(); got != 0o600 { + t.Fatalf("socket permissions = %o, want 600", got) + } + if listener.Addr().Network() != "unix" { + t.Fatalf("network = %q, want unix", listener.Addr().Network()) + } + if err := cleanup(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(socketPath); !os.IsNotExist(err) { + t.Fatalf("socket still exists after cleanup: %v", err) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 2afed6b..9cf256b 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -1,6 +1,7 @@ package api import ( + "encoding/base64" "encoding/json" "errors" "fmt" @@ -15,7 +16,7 @@ import ( "github.com/Rionlyu/spoold/internal/worker" ) -const maxRequestBody = 1 << 20 +const maxRequestBody = 2 << 20 type metricsSource interface { Metrics() worker.Metrics @@ -34,6 +35,7 @@ type createRequest struct { Method string `json:"method"` Headers map[string]string `json:"headers"` Body json.RawMessage `json:"body"` + BodyBase64 *string `json:"bodyBase64"` MaxAttempts int `json:"maxAttempts"` } @@ -73,13 +75,18 @@ func (s *Server) create(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid_request", err.Error()) return } + body, err := requestBody(payload) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid_request", err.Error()) + return + } item, created, err := s.store.Create(delivery.CreateRequest{ IdempotencyKey: payload.IdempotencyKey, TargetURL: payload.TargetURL, Method: payload.Method, Headers: payload.Headers, - Body: payload.Body, + Body: body, MaxAttempts: payload.MaxAttempts, }, time.Now()) if err != nil { @@ -108,6 +115,20 @@ func (s *Server) create(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, item) } +func requestBody(payload createRequest) ([]byte, error) { + if len(payload.Body) > 0 && payload.BodyBase64 != nil { + return nil, errors.New("body and bodyBase64 cannot be used together") + } + if payload.BodyBase64 == nil { + return append([]byte(nil), payload.Body...), nil + } + body, err := base64.StdEncoding.Strict().DecodeString(*payload.BodyBase64) + if err != nil { + return nil, errors.New("bodyBase64 must contain valid standard base64") + } + return body, nil +} + func (s *Server) list(w http.ResponseWriter, r *http.Request) { status := delivery.Status(r.URL.Query().Get("status")) if status != "" && !validStatus(status) { @@ -129,6 +150,10 @@ func (s *Server) list(w http.ResponseWriter, r *http.Request) { if len(items) > limit { items = items[:limit] } + for index := range items { + items[index].Headers = nil + items[index].Body = nil + } writeJSON(w, http.StatusOK, map[string]any{ "deliveries": items, "count": len(items), diff --git a/internal/api/server_test.go b/internal/api/server_test.go index de6c7cc..ccf94e9 100644 --- a/internal/api/server_test.go +++ b/internal/api/server_test.go @@ -65,6 +65,37 @@ func TestConflictingIdempotencyKeyReturnsConflict(t *testing.T) { } } +func TestCreateAcceptsBase64Body(t *testing.T) { + server, journal := newTestServer(t) + got := request(t, server, http.MethodPost, "/v1/deliveries", `{ + "targetUrl":"https://example.com/upload", + "bodyBase64":"AP8Q" + }`) + if got.Code != http.StatusCreated { + t.Fatalf("status = %d, body = %s", got.Code, got.Body.String()) + } + items := journal.List("") + if len(items) != 1 || !bytes.Equal(items[0].Body, []byte{0x00, 0xff, 0x10}) { + t.Fatalf("deliveries = %#v", items) + } + if !strings.Contains(got.Body.String(), `"bodyBase64":"AP8Q"`) { + t.Fatalf("response body = %s", got.Body.String()) + } +} + +func TestCreateRejectsAmbiguousOrInvalidBase64Body(t *testing.T) { + server, _ := newTestServer(t) + for _, body := range []string{ + `{"targetUrl":"https://example.com","body":{},"bodyBase64":"e30="}`, + `{"targetUrl":"https://example.com","bodyBase64":"not base64"}`, + } { + got := request(t, server, http.MethodPost, "/v1/deliveries", body) + if got.Code != http.StatusBadRequest { + t.Fatalf("status = %d, body = %s", got.Code, got.Body.String()) + } + } +} + func TestCancelDeliveryAndFilterList(t *testing.T) { server, _ := newTestServer(t) created := request(t, server, http.MethodPost, "/v1/deliveries", `{"targetUrl":"https://example.com"}`) @@ -83,6 +114,32 @@ func TestCancelDeliveryAndFilterList(t *testing.T) { } } +func TestListOmitsPayloadWhileGetReturnsIt(t *testing.T) { + server, _ := newTestServer(t) + created := request(t, server, http.MethodPost, "/v1/deliveries", `{ + "targetUrl":"https://example.com", + "headers":{"X-Secret":"value"}, + "bodyBase64":"AP8Q" + }`) + if created.Code != http.StatusCreated { + t.Fatalf("create status = %d, body = %s", created.Code, created.Body.String()) + } + var item delivery.Delivery + if err := json.Unmarshal(created.Body.Bytes(), &item); err != nil { + t.Fatal(err) + } + + list := request(t, server, http.MethodGet, "/v1/deliveries", "") + if strings.Contains(list.Body.String(), "bodyBase64") || strings.Contains(list.Body.String(), "X-Secret") { + t.Fatalf("list exposed payload: %s", list.Body.String()) + } + get := request(t, server, http.MethodGet, "/v1/deliveries/"+item.ID, "") + if !strings.Contains(get.Body.String(), `"bodyBase64":"AP8Q"`) || + !strings.Contains(get.Body.String(), `"X-Secret":"value"`) { + t.Fatalf("get omitted payload: %s", get.Body.String()) + } +} + func TestRejectsUnknownJSONField(t *testing.T) { server, _ := newTestServer(t) got := request(t, server, http.MethodPost, "/v1/deliveries", `{ @@ -111,7 +168,7 @@ func TestReadinessReflectsJournalAvailability(t *testing.T) { } func TestCreateReturnsInsufficientStorageAtAdmissionLimit(t *testing.T) { - journal, err := store.OpenWithOptions(filepath.Join(t.TempDir(), "journal"), store.Options{ + journal, err := store.Open(filepath.Join(t.TempDir(), "journal"), store.Options{ MaxJournalBytes: 1, }) if err != nil { diff --git a/internal/delivery/delivery.go b/internal/delivery/delivery.go index 33ed775..06b8901 100644 --- a/internal/delivery/delivery.go +++ b/internal/delivery/delivery.go @@ -3,10 +3,12 @@ package delivery import ( "crypto/rand" "crypto/sha256" + "encoding/base64" "encoding/hex" "encoding/json" "errors" "fmt" + "io" "net/http" "net/url" "sort" @@ -16,6 +18,8 @@ import ( type Status string +const MaxBodyBytes = 1 << 20 + const ( StatusPending Status = "pending" StatusInFlight Status = "in_flight" @@ -30,7 +34,7 @@ type Delivery struct { TargetURL string `json:"targetUrl"` Method string `json:"method"` Headers map[string]string `json:"headers,omitempty"` - Body json.RawMessage `json:"body,omitempty"` + Body []byte `json:"-"` Status Status `json:"status"` Attempts int `json:"attempts"` MaxAttempts int `json:"maxAttempts"` @@ -47,7 +51,7 @@ type CreateRequest struct { TargetURL string Method string Headers map[string]string - Body json.RawMessage + Body []byte MaxAttempts int } @@ -80,7 +84,7 @@ func New(req CreateRequest, now time.Time) (Delivery, string, error) { func Clone(d Delivery) Delivery { d.Headers = cloneHeaders(d.Headers) - d.Body = append(json.RawMessage(nil), d.Body...) + d.Body = append([]byte(nil), d.Body...) return d } @@ -88,8 +92,10 @@ func (d Delivery) MarshalJSON() ([]byte, error) { type alias Delivery value := struct { alias - NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"` - LeaseUntil *time.Time `json:"leaseUntil,omitempty"` + NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"` + LeaseUntil *time.Time `json:"leaseUntil,omitempty"` + Body json.RawMessage `json:"body,omitempty"` + BodyBase64 string `json:"bodyBase64,omitempty"` }{ alias: alias(d), } @@ -99,9 +105,43 @@ func (d Delivery) MarshalJSON() ([]byte, error) { if !d.LeaseUntil.IsZero() { value.LeaseUntil = &d.LeaseUntil } + if len(d.Body) > 0 { + if json.Valid(d.Body) { + value.Body = append(json.RawMessage(nil), d.Body...) + } else { + value.BodyBase64 = base64.StdEncoding.EncodeToString(d.Body) + } + } return json.Marshal(value) } +func (d *Delivery) UnmarshalJSON(data []byte) error { + type alias Delivery + value := struct { + alias + Body json.RawMessage `json:"body"` + BodyBase64 *string `json:"bodyBase64"` + }{} + if err := json.Unmarshal(data, &value); err != nil { + return err + } + if len(value.Body) > 0 && value.BodyBase64 != nil { + return errors.New("delivery contains both body and bodyBase64") + } + + body := append([]byte(nil), value.Body...) + if value.BodyBase64 != nil { + decoded, err := base64.StdEncoding.Strict().DecodeString(*value.BodyBase64) + if err != nil { + return fmt.Errorf("decode delivery bodyBase64: %w", err) + } + body = decoded + } + *d = Delivery(value.alias) + d.Body = body + return nil +} + func normalize(req CreateRequest) (CreateRequest, error) { req.IdempotencyKey = strings.TrimSpace(req.IdempotencyKey) if len(req.IdempotencyKey) > 256 { @@ -166,14 +206,18 @@ func normalize(req CreateRequest) (CreateRequest, error) { } req.Headers = normalizedHeaders - if len(req.Body) > 0 && !json.Valid(req.Body) { - return CreateRequest{}, errors.New("body must be valid JSON") + if len(req.Body) > MaxBodyBytes { + return CreateRequest{}, fmt.Errorf("body must not exceed %d bytes", MaxBodyBytes) } - req.Body = append(json.RawMessage(nil), req.Body...) + req.Body = append([]byte(nil), req.Body...) return req, nil } func fingerprint(req CreateRequest) string { + if len(req.Body) > 0 && !json.Valid(req.Body) { + return binaryFingerprint(req) + } + hash := sha256.New() fmt.Fprintf(hash, "%s\n%s\n%d\n", req.Method, req.TargetURL, req.MaxAttempts) names := make([]string, 0, len(req.Headers)) @@ -188,6 +232,32 @@ func fingerprint(req CreateRequest) string { return hex.EncodeToString(hash.Sum(nil)) } +func binaryFingerprint(req CreateRequest) string { + hash := sha256.New() + hash.Write([]byte("spoold-request-v2\x00")) + writeFingerprintField(hash, req.Method) + writeFingerprintField(hash, req.TargetURL) + writeFingerprintField(hash, fmt.Sprint(req.MaxAttempts)) + + names := make([]string, 0, len(req.Headers)) + for name := range req.Headers { + names = append(names, name) + } + sort.Strings(names) + fmt.Fprintf(hash, "%d:", len(names)) + for _, name := range names { + writeFingerprintField(hash, name) + writeFingerprintField(hash, req.Headers[name]) + } + fmt.Fprintf(hash, "%d:", len(req.Body)) + hash.Write(req.Body) + return hex.EncodeToString(hash.Sum(nil)) +} + +func writeFingerprintField(writer io.Writer, value string) { + fmt.Fprintf(writer, "%d:%s", len(value), value) +} + func newID() (string, error) { var raw [16]byte if _, err := rand.Read(raw[:]); err != nil { diff --git a/internal/delivery/delivery_test.go b/internal/delivery/delivery_test.go index b0812be..cbecf8d 100644 --- a/internal/delivery/delivery_test.go +++ b/internal/delivery/delivery_test.go @@ -1,6 +1,7 @@ package delivery import ( + "bytes" "encoding/json" "strings" "testing" @@ -28,10 +29,10 @@ func TestNewRejectsUnsafeOrMalformedInput(t *testing.T) { {TargetURL: "file:///tmp/out"}, {TargetURL: "https://user:pass@example.com/hook"}, {TargetURL: "https://example.com/hook#fragment"}, - {TargetURL: "https://example.com", Body: json.RawMessage(`{`)}, {TargetURL: "https://example.com", Headers: map[string]string{"X-Test": "bad\nvalue"}}, {TargetURL: "https://example.com", Headers: map[string]string{"Content-Length": "5"}}, {TargetURL: "https://example.com", Headers: map[string]string{"x-test": "a", "X-Test": "b"}}, + {TargetURL: "https://example.com", Body: make([]byte, MaxBodyBytes+1)}, } for _, request := range tests { if _, _, err := New(request, time.Now()); err == nil { @@ -40,6 +41,52 @@ func TestNewRejectsUnsafeOrMalformedInput(t *testing.T) { } } +func TestBinaryBodyUsesBase64JSONAndRoundTrips(t *testing.T) { + item, _, err := New(CreateRequest{ + TargetURL: "https://example.com/hook", + Body: []byte{0x00, 0xff, 0x10}, + }, time.Now()) + if err != nil { + t.Fatal(err) + } + data, err := json.Marshal(item) + if err != nil { + t.Fatal(err) + } + if !bytes.Contains(data, []byte(`"bodyBase64":"AP8Q"`)) { + t.Fatalf("JSON = %s", data) + } + + var decoded Delivery + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatal(err) + } + if !bytes.Equal(decoded.Body, item.Body) { + t.Fatalf("decoded body = %v, want %v", decoded.Body, item.Body) + } +} + +func TestBinaryFingerprintSeparatesHeadersFromBody(t *testing.T) { + _, firstHash, err := New(CreateRequest{ + TargetURL: "https://example.com/hook", + Body: []byte("X-Test:value\npayload"), + }, time.Now()) + if err != nil { + t.Fatal(err) + } + _, secondHash, err := New(CreateRequest{ + TargetURL: "https://example.com/hook", + Headers: map[string]string{"X-Test": "value"}, + Body: []byte("payload"), + }, time.Now()) + if err != nil { + t.Fatal(err) + } + if firstHash == secondHash { + t.Fatal("fingerprints collide across header/body boundary") + } +} + func TestMarshalOmitsInactiveScheduleFields(t *testing.T) { item := Delivery{ ID: "test", diff --git a/internal/spoolctl/spoolctl.go b/internal/spoolctl/spoolctl.go index 87d674b..e31bfe6 100644 --- a/internal/spoolctl/spoolctl.go +++ b/internal/spoolctl/spoolctl.go @@ -3,11 +3,13 @@ package spoolctl import ( "bytes" "context" + "encoding/base64" "encoding/json" "errors" "flag" "fmt" "io" + "net" "net/http" "net/url" "os" @@ -22,7 +24,6 @@ import ( const ( defaultServer = "http://127.0.0.1:8080" maxResponse = 2 << 20 - maxInputBody = 1 << 20 ) var errHelp = errors.New("help requested") @@ -41,6 +42,7 @@ type createRequest struct { Method string `json:"method,omitempty"` Headers map[string]string `json:"headers,omitempty"` Body json.RawMessage `json:"body,omitempty"` + BodyBase64 string `json:"bodyBase64,omitempty"` MaxAttempts int `json:"maxAttempts,omitempty"` } @@ -71,6 +73,7 @@ func (e apiError) Error() string { type client struct { baseURL *url.URL + displayURL string httpClient *http.Client } @@ -133,6 +136,7 @@ func runSend(ctx context.Context, args []string, stdin io.Reader, stdout, stderr maxAttempts := flags.Int("max-attempts", 0, "maximum delivery attempts (server default: 8)") data := flags.String("data", "", "JSON request body") dataFile := flags.String("data-file", "", "read JSON request body from a file, or - for stdin") + dataBinary := flags.String("data-binary", "", "read an arbitrary request body from a file, or - for stdin") jsonOutput := flags.Bool("json", false, "print the API response as JSON") headers := make(headerValues) flags.Var(&headers, "header", "outbound header in 'Name: value' form; repeatable") @@ -148,15 +152,26 @@ func runSend(ctx context.Context, args []string, stdin io.Reader, stdout, stderr flags.Usage() return usageError{errors.New("send requires exactly one target URL")} } - if *data != "" && *dataFile != "" { - return usageError{errors.New("--data and --data-file cannot be used together")} + bodySources := 0 + for _, source := range []string{*data, *dataFile, *dataBinary} { + if source != "" { + bodySources++ + } + } + if bodySources > 1 { + return usageError{errors.New("--data, --data-file, and --data-binary cannot be used together")} } - body, err := readBody(*data, *dataFile, stdin) + bodyPath := *dataFile + if *dataBinary != "" { + bodyPath = *dataBinary + } + body, err := readBody(*data, bodyPath, stdin) if err != nil { return err } - if len(body) > 0 && !json.Valid(body) { + binaryBody := *dataBinary != "" + if !binaryBody && len(body) > 0 && !json.Valid(body) { return usageError{errors.New("request body must be valid JSON")} } @@ -165,17 +180,22 @@ func runSend(ctx context.Context, args []string, stdin io.Reader, stdout, stderr return usageError{err} } var item delivery.Delivery + payload := createRequest{ + IdempotencyKey: *idempotencyKey, + TargetURL: flags.Arg(0), + Method: *method, + Headers: headers, + MaxAttempts: *maxAttempts, + } + if binaryBody { + payload.BodyBase64 = base64.StdEncoding.EncodeToString(body) + } else { + payload.Body = body + } status, err := api.do(ctx, apiRequest{ method: http.MethodPost, path: "/v1/deliveries", - body: createRequest{ - IdempotencyKey: *idempotencyKey, - TargetURL: flags.Arg(0), - Method: *method, - Headers: headers, - Body: body, - MaxAttempts: *maxAttempts, - }, + body: payload, }, &item) if err != nil { return err @@ -318,12 +338,12 @@ func readBody(inline, path string, stdin io.Reader) (json.RawMessage, error) { defer file.Close() reader = file } - body, err := io.ReadAll(io.LimitReader(reader, maxInputBody+1)) + body, err := io.ReadAll(io.LimitReader(reader, delivery.MaxBodyBytes+1)) if err != nil { return nil, fmt.Errorf("read request body: %w", err) } - if len(body) > maxInputBody { - return nil, fmt.Errorf("request body exceeds %d bytes", maxInputBody) + if len(body) > delivery.MaxBodyBytes { + return nil, fmt.Errorf("request body exceeds %d bytes", delivery.MaxBodyBytes) } return json.RawMessage(body), nil } @@ -380,12 +400,16 @@ func serverURL() string { } func newClient(rawURL string) (*client, error) { - parsed, err := url.Parse(strings.TrimSpace(rawURL)) + rawURL = strings.TrimSpace(rawURL) + parsed, err := url.Parse(rawURL) if err != nil { return nil, fmt.Errorf("parse --server: %w", err) } + if parsed.Scheme == "unix" { + return newUnixClient(rawURL, parsed) + } if parsed.Scheme != "http" && parsed.Scheme != "https" { - return nil, errors.New("--server must use http or https") + return nil, errors.New("--server must use http, https, or unix") } if parsed.Host == "" { return nil, errors.New("--server must include a host") @@ -395,7 +419,8 @@ func newClient(rawURL string) (*client, error) { } parsed.Path = strings.TrimRight(parsed.Path, "/") return &client{ - baseURL: parsed, + baseURL: parsed, + displayURL: rawURL, httpClient: &http.Client{ Timeout: 20 * time.Second, }, @@ -426,7 +451,7 @@ func (c *client) do(ctx context.Context, spec apiRequest, responseBody any) (int response, err := c.httpClient.Do(request) if err != nil { - return 0, fmt.Errorf("contact spoold at %s: %w", c.baseURL, err) + return 0, fmt.Errorf("contact spoold at %s: %w", c.displayURL, err) } defer response.Body.Close() responseData, err := io.ReadAll(io.LimitReader(response.Body, maxResponse+1)) @@ -460,6 +485,35 @@ func (c *client) do(ctx context.Context, spec apiRequest, responseBody any) (int return response.StatusCode, nil } +func newUnixClient(rawURL string, parsed *url.URL) (*client, error) { + if parsed.Host != "" || parsed.Path == "" || !strings.HasPrefix(parsed.Path, "/") { + return nil, errors.New("unix --server must contain an absolute socket path") + } + if parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" { + return nil, errors.New("unix --server must not include user information, a query, or a fragment") + } + + socketPath := parsed.Path + dialer := &net.Dialer{Timeout: 20 * time.Second} + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + transport.DialContext = func(ctx context.Context, _, _ string) (net.Conn, error) { + return dialer.DialContext(ctx, "unix", socketPath) + } + baseURL, err := url.Parse("http://spoold") + if err != nil { + return nil, err + } + return &client{ + baseURL: baseURL, + displayURL: rawURL, + httpClient: &http.Client{ + Transport: transport, + Timeout: 20 * time.Second, + }, + }, nil +} + type headerValues map[string]string func (h *headerValues) String() string { diff --git a/internal/spoolctl/spoolctl_test.go b/internal/spoolctl/spoolctl_test.go index ef6d736..4acee40 100644 --- a/internal/spoolctl/spoolctl_test.go +++ b/internal/spoolctl/spoolctl_test.go @@ -7,8 +7,10 @@ import ( "errors" "io" "log/slog" + "net" "net/http" "net/http/httptest" + "os" "path/filepath" "strings" "testing" @@ -115,6 +117,73 @@ func TestSendReadsBodyFromStdin(t *testing.T) { } } +func TestSendReadsArbitraryBinaryBody(t *testing.T) { + server, journal := newSpooldServer(t) + bodyPath := filepath.Join(t.TempDir(), "payload.bin") + want := []byte{0x00, 0xff, 0x10} + if err := os.WriteFile(bodyPath, want, 0o600); err != nil { + t.Fatal(err) + } + + var stdout, stderr bytes.Buffer + code := Run(context.Background(), []string{ + "send", + "--server", server.URL, + "--data-binary", bodyPath, + "--header", "Content-Type: application/octet-stream", + "https://example.com/upload", + }, strings.NewReader(""), &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + items := journal.List("") + if len(items) != 1 || !bytes.Equal(items[0].Body, want) { + t.Fatalf("deliveries = %#v", items) + } +} + +func TestListConnectsThroughUnixSocket(t *testing.T) { + socketDir, err := os.MkdirTemp("/tmp", "spoolctl-socket-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(socketDir) }) + socketPath := filepath.Join(socketDir, "spoold.sock") + listener, err := net.Listen("unix", socketPath) + if err != nil { + t.Fatal(err) + } + journal, err := store.Open(filepath.Join(t.TempDir(), "journal")) + if err != nil { + listener.Close() + t.Fatal(err) + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + httpServer := &http.Server{Handler: api.New(journal, nil, logger)} + go httpServer.Serve(listener) + t.Cleanup(func() { + httpServer.Close() + journal.Close() + }) + + var stdout, stderr bytes.Buffer + code := Run(context.Background(), []string{ + "list", + "--server", "unix://" + socketPath, + "--json", + }, strings.NewReader(""), &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + var response listResponse + if err := json.Unmarshal(stdout.Bytes(), &response); err != nil { + t.Fatalf("decode stdout: %v", err) + } + if response.Count != 0 { + t.Fatalf("count = %d, want 0", response.Count) + } +} + func TestGetAndCancelUseDeliveryLifecycleAPI(t *testing.T) { server, journal := newSpooldServer(t) item, _, err := journal.Create(delivery.CreateRequest{ @@ -265,10 +334,15 @@ func TestSendValidatesClientOptions(t *testing.T) { args: []string{"send", "--data", "{}", "--data-file", "-", "https://example.com"}, text: "cannot be used together", }, + { + name: "JSON and binary body sources", + args: []string{"send", "--data", "{}", "--data-binary", "-", "https://example.com"}, + text: "cannot be used together", + }, { name: "invalid server", - args: []string{"send", "--server", "unix:///tmp/spoold.sock", "https://example.com"}, - text: "must use http or https", + args: []string{"send", "--server", "smtp://localhost", "https://example.com"}, + text: "must use http, https, or unix", }, { name: "invalid header", diff --git a/internal/store/store.go b/internal/store/store.go index de1d8a6..b60d837 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -77,12 +77,15 @@ const ( compactionAfterDirSync compactionStage = "after_directory_sync" ) -func Open(path string) (*Store, error) { - return OpenWithOptions(path, Options{}) -} - -func OpenWithOptions(path string, options Options) (*Store, error) { - if options.MaxJournalBytes < 0 { +func Open(path string, options ...Options) (*Store, error) { + if len(options) > 1 { + return nil, errors.New("only one store options value is supported") + } + cfg := Options{} + if len(options) == 1 { + cfg = options[0] + } + if cfg.MaxJournalBytes < 0 { return nil, errors.New("maximum journal size must not be negative") } dir := filepath.Dir(path) @@ -130,7 +133,7 @@ func OpenWithOptions(path string, options Options) (*Store, error) { items: make(map[string]delivery.Delivery), hashes: make(map[string]string), keyToID: make(map[string]string), - maxJournalBytes: options.MaxJournalBytes, + maxJournalBytes: cfg.MaxJournalBytes, } if err := store.replay(); err != nil { file.Close() @@ -213,6 +216,15 @@ func (s *Store) List(status delivery.Status) []delivery.Delivery { } func (s *Store) ClaimDue(now time.Time, leaseDuration time.Duration, limit int) ([]delivery.Delivery, error) { + return s.ClaimDueMatching(now, leaseDuration, limit, nil) +} + +func (s *Store) ClaimDueMatching( + now time.Time, + leaseDuration time.Duration, + limit int, + eligible func(delivery.Delivery) bool, +) ([]delivery.Delivery, error) { if limit < 1 { return nil, nil } @@ -224,7 +236,7 @@ func (s *Store) ClaimDue(now time.Time, leaseDuration time.Duration, limit int) for _, item := range s.items { pending := item.Status == delivery.StatusPending && !item.NextAttemptAt.After(now) expired := item.Status == delivery.StatusInFlight && !item.LeaseUntil.After(now) - if pending || expired { + if (pending || expired) && (eligible == nil || eligible(delivery.Clone(item))) { candidates = append(candidates, item) } } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 1470bf3..a2aca7b 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -2,6 +2,7 @@ package store import ( "bufio" + "bytes" "encoding/json" "errors" "os" @@ -125,7 +126,7 @@ func TestJournalAdmissionLimitRejectsOnlyNewDeliveries(t *testing.T) { t.Fatal(err) } - bounded, err := OpenWithOptions(path, Options{MaxJournalBytes: info.Size()}) + bounded, err := Open(path, Options{MaxJournalBytes: info.Size()}) if err != nil { t.Fatal(err) } @@ -364,6 +365,41 @@ func TestReplayAcceptsCompleteFinalRecordWithoutNewline(t *testing.T) { } } +func TestBinaryBodySurvivesCompactionAndReplay(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + store, err := Open(path) + if err != nil { + t.Fatal(err) + } + want := []byte{0x00, 0xff, 0x10} + item, _, err := store.Create(delivery.CreateRequest{ + TargetURL: "https://example.com/upload", + Body: want, + }, time.Now()) + if err != nil { + t.Fatal(err) + } + if err := store.Compact(); err != nil { + t.Fatal(err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { reopened.Close() }) + got, err := reopened.Get(item.ID) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got.Body, want) { + t.Fatalf("body = %v, want %v", got.Body, want) + } +} + func TestCompactPreservesEveryDeliveryAndIdempotencyHash(t *testing.T) { path := filepath.Join(t.TempDir(), "spoold.journal") store, err := Open(path) diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 158409f..3b970c9 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -3,11 +3,13 @@ package worker import ( "bytes" "context" + "encoding/json" "fmt" "hash/fnv" "io" "log/slog" "net/http" + "net/url" "strconv" "strings" "sync" @@ -24,6 +26,7 @@ type Config struct { LeaseDuration time.Duration BaseBackoff time.Duration MaxBackoff time.Duration + PerTarget int } type Metrics struct { @@ -43,6 +46,8 @@ type Pool struct { succeeded atomic.Uint64 retryableFailure atomic.Uint64 terminalFailure atomic.Uint64 + claimMu sync.Mutex + activeTargets map[string]int wg sync.WaitGroup } @@ -62,11 +67,15 @@ func New(store *store.Store, client *http.Client, logger *slog.Logger, config Co if config.MaxBackoff <= 0 { config.MaxBackoff = 5 * time.Minute } + if config.PerTarget < 1 { + config.PerTarget = 1 + } return &Pool{ - store: store, - client: client, - log: logger, - config: config, + store: store, + client: client, + log: logger, + config: config, + activeTargets: make(map[string]int), } } @@ -102,11 +111,12 @@ func (p *Pool) run(ctx context.Context) { case <-timer.C: } - claimed, err := p.store.ClaimDue(time.Now(), p.config.LeaseDuration, 1) + claimed, err := p.claimDue(time.Now()) if err != nil { p.log.Error("claim delivery", "error", err) } else if len(claimed) == 1 { p.deliver(ctx, claimed[0]) + p.releaseTarget(claimed[0]) timer.Reset(0) continue } @@ -114,6 +124,31 @@ func (p *Pool) run(ctx context.Context) { } } +func (p *Pool) claimDue(now time.Time) ([]delivery.Delivery, error) { + p.claimMu.Lock() + defer p.claimMu.Unlock() + + claimed, err := p.store.ClaimDueMatching(now, p.config.LeaseDuration, 1, func(item delivery.Delivery) bool { + return p.activeTargets[targetKey(item.TargetURL)] < p.config.PerTarget + }) + if err == nil && len(claimed) == 1 { + p.activeTargets[targetKey(claimed[0].TargetURL)]++ + } + return claimed, err +} + +func (p *Pool) releaseTarget(item delivery.Delivery) { + p.claimMu.Lock() + defer p.claimMu.Unlock() + + key := targetKey(item.TargetURL) + if p.activeTargets[key] <= 1 { + delete(p.activeTargets, key) + return + } + p.activeTargets[key]-- +} + func (p *Pool) deliver(ctx context.Context, item delivery.Delivery) { p.attempts.Add(1) started := time.Now() @@ -158,7 +193,11 @@ func (p *Pool) send(ctx context.Context, item delivery.Delivery) (int, error) { request.Header.Set(name, value) } if len(item.Body) > 0 && request.Header.Get("Content-Type") == "" { - request.Header.Set("Content-Type", "application/json") + contentType := "application/octet-stream" + if json.Valid(item.Body) { + contentType = "application/json" + } + request.Header.Set("Content-Type", contentType) } request.Header.Set("X-Spoold-Delivery-ID", item.ID) request.Header.Set("X-Spoold-Attempt", strconv.Itoa(item.Attempts)) @@ -221,3 +260,11 @@ func failureMessage(status int, err error) string { } return fmt.Sprintf("HTTP %d", status) } + +func targetKey(targetURL string) string { + parsed, err := url.Parse(targetURL) + if err != nil { + return strings.ToLower(targetURL) + } + return strings.ToLower(parsed.Scheme + "://" + parsed.Host) +} diff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go index 37c139e..162e3cf 100644 --- a/internal/worker/worker_test.go +++ b/internal/worker/worker_test.go @@ -128,6 +128,94 @@ func TestPoolDoesNotRetryClientError(t *testing.T) { t.Fatal("delivery did not fail before timeout") } +func TestPoolDoesNotLetOneTargetConsumeAllWorkers(t *testing.T) { + blocked := make(chan struct{}) + firstTargetStarted := make(chan struct{}, 1) + firstTarget := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + select { + case firstTargetStarted <- struct{}{}: + default: + } + <-blocked + w.WriteHeader(http.StatusNoContent) + })) + defer firstTarget.Close() + + secondTargetCalled := make(chan struct{}, 1) + secondTarget := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + secondTargetCalled <- struct{}{} + w.WriteHeader(http.StatusNoContent) + })) + defer secondTarget.Close() + + journal, err := store.Open(filepath.Join(t.TempDir(), "journal")) + if err != nil { + t.Fatal(err) + } + defer journal.Close() + now := time.Now() + for index, targetURL := range []string{firstTarget.URL, firstTarget.URL, secondTarget.URL} { + if _, _, err := journal.Create(delivery.CreateRequest{ + TargetURL: targetURL, + }, now.Add(time.Duration(index)*time.Millisecond)); err != nil { + t.Fatal(err) + } + } + + pool := New(journal, target.NewClient(true, time.Second), slog.New(slog.NewTextHandler(io.Discard, nil)), Config{ + Concurrency: 2, + PerTarget: 1, + PollInterval: time.Millisecond, + }) + ctx, cancel := context.WithCancel(context.Background()) + pool.Start(ctx) + defer func() { + close(blocked) + cancel() + pool.Wait() + }() + + select { + case <-firstTargetStarted: + case <-time.After(time.Second): + t.Fatal("first target was not called") + } + select { + case <-secondTargetCalled: + case <-time.After(time.Second): + t.Fatal("second target was starved by the first target") + } +} + +func TestSendDefaultsBinaryBodyContentType(t *testing.T) { + destination := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Content-Type"); got != "application/octet-stream" { + t.Errorf("Content-Type = %q, want application/octet-stream", got) + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + } + if string(body) != "\x00\xff" { + t.Errorf("body = %v", body) + } + w.WriteHeader(http.StatusNoContent) + })) + defer destination.Close() + + pool := New(nil, target.NewClient(true, time.Second), slog.New(slog.NewTextHandler(io.Discard, nil)), Config{}) + status, err := pool.send(context.Background(), delivery.Delivery{ + ID: "binary", + TargetURL: destination.URL, + Method: http.MethodPost, + Body: []byte{0x00, 0xff}, + Attempts: 1, + }) + if err != nil || status != http.StatusNoContent { + t.Fatalf("send() = (%d, %v)", status, err) + } +} + func TestBackoffIsBounded(t *testing.T) { for attempt := 1; attempt <= 20; attempt++ { got := Backoff(time.Second, 10*time.Second, "delivery", attempt) From fc59156b05e0337fead7e0939632b4bb7364b20b Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 11:57:48 +0200 Subject: [PATCH 3/9] test: prove recovery across process death --- cmd/spoold/process_test.go | 240 +++++++++++++++++++++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 cmd/spoold/process_test.go diff --git a/cmd/spoold/process_test.go b/cmd/spoold/process_test.go new file mode 100644 index 0000000..0090b4e --- /dev/null +++ b/cmd/spoold/process_test.go @@ -0,0 +1,240 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Rionlyu/spoold/internal/delivery" +) + +func TestProcessRecoversDeliveryAfterSIGKILL(t *testing.T) { + apiAddress := unusedTCPAddress(t) + targetAddress := unusedTCPAddress(t) + journalPath := filepath.Join(t.TempDir(), "spoold.journal") + args := []string{ + "-listen", apiAddress, + "-journal", journalPath, + "-allow-private-targets", + "-workers", "1", + "-per-target-workers", "1", + "-request-timeout", "100ms", + "-compact-threshold-bytes", "0", + "-terminal-retention", "0", + } + + first := startSpooldProcess(t, args) + waitForHealth(t, apiAddress, first) + item := enqueueDelivery(t, apiAddress, "http://"+targetAddress+"/events") + waitForDelivery(t, apiAddress, item.ID, first, func(item delivery.Delivery) bool { + return item.Status == delivery.StatusPending && item.Attempts >= 1 + }) + + conflictArgs := append([]string(nil), args...) + conflictArgs[1] = unusedTCPAddress(t) + conflict := newSpooldProcess(conflictArgs) + conflictOutput, err := conflict.CombinedOutput() + if err == nil { + t.Fatal("second process unexpectedly opened the same journal") + } + if !strings.Contains(string(conflictOutput), "already owned") { + t.Fatalf("second process output = %s", conflictOutput) + } + + if err := first.cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + if err := first.cmd.Wait(); err == nil { + t.Fatal("SIGKILL process exited successfully") + } + first.finished = true + + var received atomic.Int32 + targetListener, err := net.Listen("tcp", targetAddress) + if err != nil { + t.Fatal(err) + } + targetServer := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received.Add(1) + w.WriteHeader(http.StatusNoContent) + })} + go targetServer.Serve(targetListener) + t.Cleanup(func() { targetServer.Close() }) + + second := startSpooldProcess(t, args) + waitForHealth(t, apiAddress, second) + succeeded := waitForDelivery(t, apiAddress, item.ID, second, func(item delivery.Delivery) bool { + return item.Status == delivery.StatusSucceeded + }) + if succeeded.Attempts < 2 { + t.Fatalf("attempts after recovery = %d, want at least 2", succeeded.Attempts) + } + if received.Load() != 1 { + t.Fatalf("destination requests = %d, want 1", received.Load()) + } + stopSpooldProcess(t, second) +} + +func TestSpooldHelperProcess(t *testing.T) { + if os.Getenv("SPOOLD_TEST_HELPER") != "1" { + return + } + separator := -1 + for index, argument := range os.Args { + if argument == "--" { + separator = index + break + } + } + if separator == -1 { + os.Exit(2) + } + os.Exit(run(os.Args[separator+1:], os.Stdout, os.Stderr)) +} + +type spooldProcess struct { + cmd *exec.Cmd + output lockedBuffer + finished bool +} + +type lockedBuffer struct { + mu sync.Mutex + bytes.Buffer +} + +func (b *lockedBuffer) Write(data []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.Buffer.Write(data) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.Buffer.String() +} + +func newSpooldProcess(args []string) *exec.Cmd { + commandArgs := append([]string{"-test.run=^TestSpooldHelperProcess$", "--"}, args...) + cmd := exec.Command(os.Args[0], commandArgs...) + cmd.Env = append(os.Environ(), "SPOOLD_TEST_HELPER=1") + return cmd +} + +func startSpooldProcess(t *testing.T, args []string) *spooldProcess { + t.Helper() + process := &spooldProcess{cmd: newSpooldProcess(args)} + process.cmd.Stdout = &process.output + process.cmd.Stderr = &process.output + if err := process.cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if process.finished { + return + } + _ = process.cmd.Process.Kill() + _ = process.cmd.Wait() + }) + return process +} + +func stopSpooldProcess(t *testing.T, process *spooldProcess) { + t.Helper() + if err := process.cmd.Process.Signal(os.Interrupt); err != nil { + t.Fatal(err) + } + if err := process.cmd.Wait(); err != nil { + t.Fatalf("stop spoold: %v\n%s", err, process.output.String()) + } + process.finished = true +} + +func unusedTCPAddress(t *testing.T) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + address := listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatal(err) + } + return address +} + +func waitForHealth(t *testing.T, address string, process *spooldProcess) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + response, err := http.Get("http://" + address + "/healthz") + if err == nil { + response.Body.Close() + if response.StatusCode == http.StatusOK { + return + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("spoold did not become healthy\n%s", process.output.String()) +} + +func enqueueDelivery(t *testing.T, address, targetURL string) delivery.Delivery { + t.Helper() + payload := fmt.Sprintf(`{"targetUrl":%q,"body":{"crash":"recovery"}}`, targetURL) + response, err := http.Post( + "http://"+address+"/v1/deliveries", + "application/json", + strings.NewReader(payload), + ) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(response.Body) + t.Fatalf("enqueue status = %d, body = %s", response.StatusCode, body) + } + var item delivery.Delivery + if err := json.NewDecoder(response.Body).Decode(&item); err != nil { + t.Fatal(err) + } + return item +} + +func waitForDelivery( + t *testing.T, + address string, + id string, + process *spooldProcess, + done func(delivery.Delivery) bool, +) delivery.Delivery { + t.Helper() + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + response, err := http.Get("http://" + address + "/v1/deliveries/" + id) + if err == nil { + var item delivery.Delivery + decodeErr := json.NewDecoder(response.Body).Decode(&item) + response.Body.Close() + if response.StatusCode == http.StatusOK && decodeErr == nil && done(item) { + return item + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("delivery did not reach expected state\n%s", process.output.String()) + return delivery.Delivery{} +} From d0e913e995c45e20d4bcdc4d402b5055643f4563 Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 12:02:22 +0200 Subject: [PATCH 4/9] runtime: repair unhealthy journal automatically --- internal/compactor/compactor.go | 9 +++++++- internal/compactor/compactor_test.go | 31 ++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/internal/compactor/compactor.go b/internal/compactor/compactor.go index cbe60c6..1f6147d 100644 --- a/internal/compactor/compactor.go +++ b/internal/compactor/compactor.go @@ -23,6 +23,7 @@ type Config struct { type journalStore interface { Stats() store.Stats + Ready() error Compact() error PruneTerminal(time.Time) (int, error) } @@ -77,8 +78,14 @@ func (c *Compactor) run(ctx context.Context) { } func (c *Compactor) maintain() { + started := time.Now() + if err := c.store.Ready(); err != nil { + c.log.Warn("journal persistence unhealthy; attempting repair", "error", err) + _ = c.compact("persistence_repair", started) + return + } + if c.retention > 0 { - started := time.Now() pruned, err := c.store.PruneTerminal(started.Add(-c.retention)) if err != nil { c.log.Warn("journal retention failed", "error", err) diff --git a/internal/compactor/compactor_test.go b/internal/compactor/compactor_test.go index f6f5ffa..a99869b 100644 --- a/internal/compactor/compactor_test.go +++ b/internal/compactor/compactor_test.go @@ -147,6 +147,29 @@ func TestCompactionFailureIsNonFatal(t *testing.T) { compactor.Wait() } +func TestMaintenanceRepairsUnreadyJournalBelowThreshold(t *testing.T) { + journal := &fakeStore{ + stats: store.Stats{ + JournalSizeBytes: 10, + JournalRecords: 1, + LiveDeliveries: 1, + }, + readyErr: errors.New("injected persistence failure"), + } + compactor := New(journal, discardLogger(), Config{ + ThresholdBytes: 100, + }) + + compactor.maintain() + + if got := journal.compactCalls(); got != 1 { + t.Fatalf("Compact() calls = %d, want 1", got) + } + if err := journal.Ready(); err != nil { + t.Fatalf("Ready() after repair = %v, want nil", err) + } +} + func TestRetentionPrunesTerminalDeliveriesAndCompactsTombstones(t *testing.T) { now := time.Now() journal := &fakeStore{ @@ -181,6 +204,7 @@ type fakeStore struct { calls int failures int items []time.Time + readyErr error } func (s *fakeStore) Stats() store.Stats { @@ -189,6 +213,12 @@ func (s *fakeStore) Stats() store.Stats { return s.stats } +func (s *fakeStore) Ready() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.readyErr +} + func (s *fakeStore) Compact() error { s.mu.Lock() defer s.mu.Unlock() @@ -201,6 +231,7 @@ func (s *fakeStore) Compact() error { s.stats.JournalSizeBytes /= 2 s.stats.JournalRecords = s.stats.LiveDeliveries s.stats.CompactionsSucceeded++ + s.readyErr = nil return nil } From 929dd2cc7bd3f2214bcea50bdf1e1a3e35a51461 Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 12:05:40 +0200 Subject: [PATCH 5/9] worker: normalize target concurrency origins --- internal/worker/worker.go | 13 ++++++++++++- internal/worker/worker_test.go | 17 +++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 3b970c9..ab38358 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -8,6 +8,7 @@ import ( "hash/fnv" "io" "log/slog" + "net" "net/http" "net/url" "strconv" @@ -266,5 +267,15 @@ func targetKey(targetURL string) string { if err != nil { return strings.ToLower(targetURL) } - return strings.ToLower(parsed.Scheme + "://" + parsed.Host) + scheme := strings.ToLower(parsed.Scheme) + port := parsed.Port() + if port == "" { + switch scheme { + case "http": + port = "80" + case "https": + port = "443" + } + } + return scheme + "://" + net.JoinHostPort(strings.ToLower(parsed.Hostname()), port) } diff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go index 162e3cf..93ecb9b 100644 --- a/internal/worker/worker_test.go +++ b/internal/worker/worker_test.go @@ -224,3 +224,20 @@ func TestBackoffIsBounded(t *testing.T) { } } } + +func TestTargetKeyNormalizesEquivalentOrigins(t *testing.T) { + tests := []struct { + target string + want string + }{ + {target: "https://EXAMPLE.com/path", want: "https://example.com:443"}, + {target: "https://example.com:443/other", want: "https://example.com:443"}, + {target: "http://example.com", want: "http://example.com:80"}, + {target: "http://[2001:db8::1]/events", want: "http://[2001:db8::1]:80"}, + } + for _, test := range tests { + if got := targetKey(test.target); got != test.want { + t.Errorf("targetKey(%q) = %q, want %q", test.target, got, test.want) + } + } +} From 5fb9a0f31c9bff8173fb41ae8280aac90041e2df Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 12:07:35 +0200 Subject: [PATCH 6/9] store: preserve durability after replacement failure --- internal/store/store.go | 12 +++- internal/store/store_test.go | 112 ++++++++++++++++++++++++----------- 2 files changed, 86 insertions(+), 38 deletions(-) diff --git a/internal/store/store.go b/internal/store/store.go index b60d837..098c29d 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -491,23 +491,31 @@ func (s *Store) Compact() (err error) { } renamed = true if err := s.runCompactionHook(compactionAfterRename); err != nil { - err = errors.Join(err, s.adoptCompactedLocked(temp, uint64(len(ids)))) + err = s.failPersistenceLocked(errors.Join( + err, + s.adoptCompactedLocked(temp, uint64(len(ids))), + )) adopted = true return err } if err := syncDirectory(filepath.Dir(s.path)); err != nil { - err = errors.Join(err, s.adoptCompactedLocked(temp, uint64(len(ids)))) + err = s.failPersistenceLocked(errors.Join( + err, + s.adoptCompactedLocked(temp, uint64(len(ids))), + )) adopted = true return err } if err := s.runCompactionHook(compactionAfterDirSync); err != nil { err = errors.Join(err, s.adoptCompactedLocked(temp, uint64(len(ids)))) adopted = true + s.persistenceErr = nil return err } if err := s.adoptCompactedLocked(temp, uint64(len(ids))); err != nil { adopted = true + s.persistenceErr = nil return err } adopted = true diff --git a/internal/store/store_test.go b/internal/store/store_test.go index a2aca7b..dce129e 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -520,44 +520,67 @@ func TestCompactFailureBeforeRenameLeavesOriginalJournalWritable(t *testing.T) { assertReplaysIDs(t, path, first.ID, second.ID) } -func TestCompactFailureAfterReplacementKeepsSubsequentAppendsRecoverable(t *testing.T) { - for _, failureStage := range []compactionStage{compactionAfterRename, compactionAfterDirSync} { - t.Run(string(failureStage), func(t *testing.T) { - path := filepath.Join(t.TempDir(), "spoold.journal") - store, err := Open(path) - if err != nil { - t.Fatal(err) - } - first := createTestDelivery(t, store, "first", time.Now()) - claimed, err := store.ClaimDue(time.Now().Add(time.Second), time.Minute, 1) - if err != nil || len(claimed) != 1 { - t.Fatalf("ClaimDue() = %#v, %v", claimed, err) - } - if err := store.Succeed(first.ID, claimed[0].Attempts, 200, time.Now().Add(2*time.Second)); err != nil { - t.Fatal(err) - } +func TestCompactFailureAfterRenameRequiresDurabilityRepair(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + store, first := storeWithTransitionedDelivery(t, path) + store.compactionHook = func(stage compactionStage) error { + if stage == compactionAfterRename { + return errors.New("injected failure") + } + return nil + } + if err := store.Compact(); !errors.Is(err, ErrPersistence) { + t.Fatalf("Compact() error = %v, want %v", err, ErrPersistence) + } + if err := store.Ready(); !errors.Is(err, ErrPersistence) { + t.Fatalf("Ready() error = %v, want %v", err, ErrPersistence) + } + if _, _, err := store.Create(delivery.CreateRequest{ + TargetURL: "https://example.com/blocked", + }, time.Now()); !errors.Is(err, ErrPersistence) { + t.Fatalf("Create() before repair error = %v, want %v", err, ErrPersistence) + } - store.compactionHook = func(stage compactionStage) error { - if stage == failureStage { - return errors.New("injected failure") - } - return nil - } - if err := store.Compact(); err == nil { - t.Fatal("Compact() error = nil, want injected failure") - } - store.compactionHook = nil - second := createTestDelivery(t, store, "second", time.Now().Add(3*time.Second)) - stats := store.Stats() - if stats.CompactionsFailed != 1 || stats.CompactionsSucceeded != 0 || stats.JournalRecords != 2 { - t.Fatalf("stats = %#v", stats) - } - if err := store.Close(); err != nil { - t.Fatal(err) - } - assertReplaysIDs(t, path, first.ID, second.ID) - }) + store.compactionHook = nil + if err := store.Compact(); err != nil { + t.Fatalf("repairing Compact(): %v", err) + } + second := createTestDelivery(t, store, "second", time.Now().Add(3*time.Second)) + stats := store.Stats() + if stats.CompactionsFailed != 1 || stats.CompactionsSucceeded != 1 || stats.JournalRecords != 2 { + t.Fatalf("stats = %#v", stats) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + assertReplaysIDs(t, path, first.ID, second.ID) +} + +func TestCompactFailureAfterDirectorySyncKeepsSubsequentAppendsRecoverable(t *testing.T) { + path := filepath.Join(t.TempDir(), "spoold.journal") + store, first := storeWithTransitionedDelivery(t, path) + store.compactionHook = func(stage compactionStage) error { + if stage == compactionAfterDirSync { + return errors.New("injected failure") + } + return nil + } + if err := store.Compact(); err == nil { + t.Fatal("Compact() error = nil, want injected failure") } + if err := store.Ready(); err != nil { + t.Fatalf("Ready() after durable replacement = %v", err) + } + store.compactionHook = nil + second := createTestDelivery(t, store, "second", time.Now().Add(3*time.Second)) + stats := store.Stats() + if stats.CompactionsFailed != 1 || stats.CompactionsSucceeded != 0 || stats.JournalRecords != 2 { + t.Fatalf("stats = %#v", stats) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + assertReplaysIDs(t, path, first.ID, second.ID) } func TestOpenRemovesAbandonedCompactionFiles(t *testing.T) { @@ -670,6 +693,23 @@ func createTestDelivery(t *testing.T, store *Store, key string, now time.Time) d return item } +func storeWithTransitionedDelivery(t *testing.T, path string) (*Store, delivery.Delivery) { + t.Helper() + store, err := Open(path) + if err != nil { + t.Fatal(err) + } + first := createTestDelivery(t, store, "first", time.Now()) + claimed, err := store.ClaimDue(time.Now().Add(time.Second), time.Minute, 1) + if err != nil || len(claimed) != 1 { + t.Fatalf("ClaimDue() = %#v, %v", claimed, err) + } + if err := store.Succeed(first.ID, claimed[0].Attempts, 200, time.Now().Add(2*time.Second)); err != nil { + t.Fatal(err) + } + return store, first +} + func assertJournalIDsSorted(t *testing.T, path string) { t.Helper() file, err := os.Open(path) From ddd67fec6401808391b4930bfb6b2c89d8b4b4bf Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 12:12:45 +0200 Subject: [PATCH 7/9] release: automate verifiable alpha artifacts --- .dockerignore | 1 + .github/workflows/ci.yml | 89 +++++++++++++++++- .github/workflows/release.yml | 50 ++++++++++ .gitignore | 1 + .goreleaser.Dockerfile | 14 +++ .goreleaser.yml | 133 +++++++++++++++++++++++++++ Dockerfile | 19 +++- cmd/spoold/main.go | 10 ++ cmd/spoold/main_test.go | 10 ++ internal/buildinfo/buildinfo.go | 13 +++ internal/buildinfo/buildinfo_test.go | 15 +++ internal/spoolctl/spoolctl.go | 5 + internal/spoolctl/spoolctl_test.go | 1 + scripts/install.sh | 80 ++++++++++++++++ 14 files changed, 437 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/release.yml create mode 100644 .goreleaser.Dockerfile create mode 100644 .goreleaser.yml create mode 100644 internal/buildinfo/buildinfo.go create mode 100644 internal/buildinfo/buildinfo_test.go create mode 100755 scripts/install.sh diff --git a/.dockerignore b/.dockerignore index a251155..138ad52 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,6 +1,7 @@ .git .github bin +dist data/* !data/.keep *.out diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5231ee2..6236f67 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,9 +12,94 @@ jobs: verify: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v6 - - uses: actions/setup-go@v6 + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false + - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6.4.0 with: go-version: "1.26.x" cache: false - run: make verify + + package: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + fetch-depth: 0 + persist-credentials: false + - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6.4.0 + with: + go-version-file: go.mod + cache: false + - uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0 + - uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0 + - uses: goreleaser/goreleaser-action@f06c13b6b1a9625abc9e6e439d9c05a8f2190e94 # v7.2.3 + with: + version: v2.17.0 + args: release --snapshot --clean --skip=sbom,sign + - name: Verify release archives + run: | + test "$(find dist -maxdepth 1 -name '*.tar.gz' | wc -l)" -eq 5 + archive="$(find dist -maxdepth 1 -name '*_linux_amd64.tar.gz')" + tar -tzf "$archive" | grep -q '/spoold$' + tar -tzf "$archive" | grep -q '/spoolctl$' + (cd dist && sha256sum --check checksums.txt) + unpacked="$(mktemp -d)" + tar -xzf "$archive" -C "$unpacked" + "$(find "$unpacked" -type f -name spoold)" -version | grep -q 'spoold 0.0.0-SNAPSHOT' + "$(find "$unpacked" -type f -name spoolctl)" version | grep -q 'spoolctl 0.0.0-SNAPSHOT' + - name: Verify installer + run: | + installed="$(mktemp -d)" + version="$(jq -r .version dist/metadata.json)" + SPOOLD_INSTALL_DIR="$installed" \ + SPOOLD_VERSION="$version" \ + SPOOLD_RELEASE_BASE_URL="file://${GITHUB_WORKSPACE}/dist" \ + sh scripts/install.sh + "$installed/spoold" -version | grep -q "spoold ${version}" + "$installed/spoolctl" version | grep -q "spoolctl ${version}" + - name: Verify release container + run: | + image="$(docker image ls ghcr.io/rionlyu/spoold --format '{{.Repository}}:{{.Tag}}' | grep -- '-amd64$' | head -n 1)" + test -n "$image" + docker run --rm "$image" -version | grep -q 'spoold 0.0.0-SNAPSHOT' + docker volume create spoold-ci-data + trap 'docker rm -f spoold-release-ci >/dev/null 2>&1 || true; docker volume rm spoold-ci-data >/dev/null 2>&1 || true' EXIT + docker run -d \ + --name spoold-release-ci \ + --publish 127.0.0.1:18080:8080 \ + --volume spoold-ci-data:/var/lib/spoold \ + "$image" + for _ in {1..50}; do + curl -fsS http://127.0.0.1:18080/readyz && break + sleep 0.1 + done + response="$(curl -fsS \ + --header 'Content-Type: application/json' \ + --data '{"targetUrl":"http://127.0.0.1:1/events","body":{"container":"replay"}}' \ + http://127.0.0.1:18080/v1/deliveries)" + id="$(jq -r .id <<<"$response")" + test -n "$id" + test "$id" != null + docker kill --signal KILL spoold-release-ci + docker rm spoold-release-ci + docker run -d \ + --name spoold-release-ci \ + --publish 127.0.0.1:18080:8080 \ + --volume spoold-ci-data:/var/lib/spoold \ + "$image" + for _ in {1..50}; do + curl -fsS http://127.0.0.1:18080/readyz && break + sleep 0.1 + done + curl -fsS "http://127.0.0.1:18080/v1/deliveries/${id}" | + jq -e --arg id "$id" '.id == $id' + - name: Build container + run: | + docker build \ + --build-arg VERSION=ci \ + --build-arg COMMIT="${GITHUB_SHA}" \ + --build-arg DATE=1970-01-01T00:00:00Z \ + --tag spoold:ci . + docker run --rm spoold:ci -version | grep -q 'spoold ci' diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..03b47e3 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,50 @@ +name: Release + +on: + push: + tags: + - "v*" + +permissions: + contents: write + id-token: write + packages: write + attestations: write + +jobs: + release: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + fetch-depth: 0 + persist-credentials: false + - name: Require a commit merged to main + run: | + git fetch --no-tags origin main + git merge-base --is-ancestor "$GITHUB_SHA" origin/main + - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6.4.0 + with: + go-version-file: go.mod + cache: false + - uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0 + - uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0 + - uses: sigstore/cosign-installer@6f9f17788090df1f26f669e9d70d6ae9567deba6 # v4.1.2 + - uses: anchore/sbom-action/download-syft@e22c389904149dbc22b58101806040fa8d37a610 # v0.24.0 + - uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0 + with: + registry: ghcr.io + username: ${{ github.repository_owner }} + password: ${{ secrets.GITHUB_TOKEN }} + - uses: goreleaser/goreleaser-action@f06c13b6b1a9625abc9e6e439d9c05a8f2190e94 # v7.2.3 + with: + version: v2.17.0 + args: release --clean + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + - uses: actions/attest-build-provenance@0f67c3f4856b2e3261c31976d6725780e5e4c373 # v4.1.1 + with: + subject-checksums: ./dist/checksums.txt + - uses: actions/attest-build-provenance@0f67c3f4856b2e3261c31976d6725780e5e4c373 # v4.1.1 + with: + subject-checksums: ./dist/digests.txt diff --git a/.gitignore b/.gitignore index c4346f9..d02afa3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ /bin/ +/dist/ /data/* !/data/.keep *.out diff --git a/.goreleaser.Dockerfile b/.goreleaser.Dockerfile new file mode 100644 index 0000000..293ca29 --- /dev/null +++ b/.goreleaser.Dockerfile @@ -0,0 +1,14 @@ +FROM alpine:3.23 + +ARG TARGETPLATFORM + +RUN apk add --no-cache ca-certificates \ + && mkdir -p /var/lib/spoold \ + && chown 65532:65532 /var/lib/spoold + +COPY $TARGETPLATFORM/spoold /usr/local/bin/spoold + +USER 65532:65532 +EXPOSE 8080 +ENTRYPOINT ["/usr/local/bin/spoold"] +CMD ["-listen", "0.0.0.0:8080", "-journal", "/var/lib/spoold/spoold.journal"] diff --git a/.goreleaser.yml b/.goreleaser.yml new file mode 100644 index 0000000..98a9157 --- /dev/null +++ b/.goreleaser.yml @@ -0,0 +1,133 @@ +version: 2 + +project_name: spoold + +gomod: + proxy: true + +builds: + - id: spoold + main: ./cmd/spoold + binary: spoold + env: + - CGO_ENABLED=0 + goos: + - linux + - darwin + goarch: + - amd64 + - arm64 + mod_timestamp: "{{ .CommitTimestamp }}" + flags: + - -trimpath + ldflags: + - >- + -s -w + -X github.com/Rionlyu/spoold/internal/buildinfo.Version={{ .Version }} + -X github.com/Rionlyu/spoold/internal/buildinfo.Commit={{ .Commit }} + -X github.com/Rionlyu/spoold/internal/buildinfo.Date={{ .CommitDate }} + - id: spoolctl + main: ./cmd/spoolctl + binary: spoolctl + env: + - CGO_ENABLED=0 + goos: + - linux + - darwin + goarch: + - amd64 + - arm64 + mod_timestamp: "{{ .CommitTimestamp }}" + flags: + - -trimpath + ldflags: + - >- + -s -w + -X github.com/Rionlyu/spoold/internal/buildinfo.Version={{ .Version }} + -X github.com/Rionlyu/spoold/internal/buildinfo.Commit={{ .Commit }} + -X github.com/Rionlyu/spoold/internal/buildinfo.Date={{ .CommitDate }} + +source: + enabled: true + name_template: "{{ .ProjectName }}_{{ .Version }}_source" + +archives: + - ids: + - spoold + - spoolctl + formats: + - tar.gz + name_template: "{{ .ProjectName }}_{{ .Version }}_{{ .Os }}_{{ .Arch }}" + wrap_in_directory: true + builds_info: + mtime: "{{ .CommitTimestamp }}" + files: + - LICENSE + - README.md + - SECURITY.md + - CONTRIBUTING.md + - CODE_OF_CONDUCT.md + - docs/design.md + - docs/product-direction.md + - docs/launch-readiness.md + +sboms: + - artifacts: archive + - id: source + artifacts: source + +checksum: + name_template: checksums.txt + +signs: + - cmd: cosign + signature: "${artifact}.sigstore.json" + args: + - sign-blob + - "--bundle=${signature}" + - "${artifact}" + - "--yes" + artifacts: checksum + output: true + +dockers_v2: + - id: spoold + ids: + - spoold + images: + - ghcr.io/rionlyu/spoold + tags: + - "{{ .Tag }}" + - "{{ if not .IsPrerelease }}latest{{ end }}" + dockerfile: .goreleaser.Dockerfile + platforms: + - linux/amd64 + - linux/arm64 + labels: + org.opencontainers.image.created: "{{ .Date }}" + org.opencontainers.image.name: "{{ .ProjectName }}" + org.opencontainers.image.revision: "{{ .FullCommit }}" + org.opencontainers.image.version: "{{ .Version }}" + org.opencontainers.image.source: "{{ .GitURL }}" + org.opencontainers.image.description: "Crash-safe local HTTP delivery spool" + org.opencontainers.image.licenses: MIT + annotations: + org.opencontainers.image.description: "Crash-safe local HTTP delivery spool" + +docker_signs: + - cmd: cosign + output: true + args: + - sign + - "${artifact}" + - "--yes" + +report_sizes: true + +changelog: + sort: asc + filters: + exclude: + - "^docs:" + - "^test:" + - "^chore:" diff --git a/Dockerfile b/Dockerfile index cc4c589..30eb25e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,19 +1,34 @@ FROM golang:1.26-alpine AS build +ARG VERSION=dev +ARG COMMIT=unknown +ARG DATE=unknown + WORKDIR /src COPY go.mod ./ COPY cmd ./cmd COPY internal ./internal -RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /spoold ./cmd/spoold +RUN CGO_ENABLED=0 go build -trimpath \ + -ldflags="-s -w \ + -X github.com/Rionlyu/spoold/internal/buildinfo.Version=${VERSION} \ + -X github.com/Rionlyu/spoold/internal/buildinfo.Commit=${COMMIT} \ + -X github.com/Rionlyu/spoold/internal/buildinfo.Date=${DATE}" \ + -o /spoold ./cmd/spoold FROM scratch +ARG VERSION=dev + COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt COPY --from=build --chown=65532:65532 /spoold /spoold COPY --chown=65532:65532 data/.keep /var/lib/spoold/.keep +LABEL org.opencontainers.image.source="https://github.com/Rionlyu/spoold" \ + org.opencontainers.image.description="Crash-safe local HTTP delivery spool" \ + org.opencontainers.image.licenses="MIT" \ + org.opencontainers.image.version="${VERSION}" + USER 65532:65532 EXPOSE 8080 ENTRYPOINT ["/spoold"] CMD ["-listen", "0.0.0.0:8080", "-journal", "/var/lib/spoold/spoold.journal"] - diff --git a/cmd/spoold/main.go b/cmd/spoold/main.go index 45ef855..11182bd 100644 --- a/cmd/spoold/main.go +++ b/cmd/spoold/main.go @@ -16,6 +16,7 @@ import ( "time" "github.com/Rionlyu/spoold/internal/api" + "github.com/Rionlyu/spoold/internal/buildinfo" "github.com/Rionlyu/spoold/internal/compactor" "github.com/Rionlyu/spoold/internal/store" "github.com/Rionlyu/spoold/internal/target" @@ -37,6 +38,7 @@ type config struct { shutdownTimeout time.Duration compactThreshold int64 compactInterval time.Duration + showVersion bool } func main() { @@ -48,6 +50,10 @@ func run(args []string, stdout, stderr io.Writer) int { if code != 0 { return code } + if cfg.showVersion { + fmt.Fprintf(stdout, "spoold %s\n", buildinfo.String()) + return 0 + } logger := slog.New(slog.NewJSONHandler(stdout, nil)) journal, err := store.Open(cfg.journalPath, store.Options{ @@ -155,6 +161,7 @@ func parseConfig(args []string, stderr io.Writer) (config, int) { flags.DurationVar(&cfg.shutdownTimeout, "shutdown-timeout", 10*time.Second, "graceful shutdown timeout") flags.Int64Var(&cfg.compactThreshold, "compact-threshold-bytes", compactor.DefaultThresholdBytes, "minimum journal size for compaction (0 disables size-based compaction)") flags.DurationVar(&cfg.compactInterval, "compact-check-interval", compactor.DefaultCheckInterval, "journal maintenance interval") + flags.BoolVar(&cfg.showVersion, "version", false, "print version information") if err := flags.Parse(args); err != nil { return config{}, 2 } @@ -162,6 +169,9 @@ func parseConfig(args []string, stderr io.Writer) (config, int) { fmt.Fprintln(stderr, "spoold: positional arguments are not supported") return config{}, 2 } + if cfg.showVersion { + return cfg, 0 + } switch { case cfg.journalPath == "": diff --git a/cmd/spoold/main_test.go b/cmd/spoold/main_test.go index c956980..1ca8efd 100644 --- a/cmd/spoold/main_test.go +++ b/cmd/spoold/main_test.go @@ -29,6 +29,16 @@ func TestRunReturnsFailureWhenAddressIsAlreadyInUse(t *testing.T) { } } +func TestRunPrintsVersionWithoutStarting(t *testing.T) { + var stdout, stderr bytes.Buffer + if code := run([]string{"-version"}, &stdout, &stderr); code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + if !strings.Contains(stdout.String(), "spoold dev") { + t.Fatalf("stdout = %q", stdout.String()) + } +} + func TestParseConfigRejectsUnsafeValues(t *testing.T) { tests := [][]string{ {"-workers", "0"}, diff --git a/internal/buildinfo/buildinfo.go b/internal/buildinfo/buildinfo.go new file mode 100644 index 0000000..0e9a45d --- /dev/null +++ b/internal/buildinfo/buildinfo.go @@ -0,0 +1,13 @@ +package buildinfo + +import "fmt" + +var ( + Version = "dev" + Commit = "unknown" + Date = "unknown" +) + +func String() string { + return fmt.Sprintf("%s (commit %s, built %s)", Version, Commit, Date) +} diff --git a/internal/buildinfo/buildinfo_test.go b/internal/buildinfo/buildinfo_test.go new file mode 100644 index 0000000..80dd62f --- /dev/null +++ b/internal/buildinfo/buildinfo_test.go @@ -0,0 +1,15 @@ +package buildinfo + +import ( + "strings" + "testing" +) + +func TestStringIncludesBuildFields(t *testing.T) { + got := String() + for _, value := range []string{Version, Commit, Date} { + if !strings.Contains(got, value) { + t.Fatalf("String() = %q, want %q", got, value) + } + } +} diff --git a/internal/spoolctl/spoolctl.go b/internal/spoolctl/spoolctl.go index e31bfe6..ea10515 100644 --- a/internal/spoolctl/spoolctl.go +++ b/internal/spoolctl/spoolctl.go @@ -18,6 +18,7 @@ import ( "text/tabwriter" "time" + "github.com/Rionlyu/spoold/internal/buildinfo" "github.com/Rionlyu/spoold/internal/delivery" ) @@ -103,6 +104,9 @@ func Run(ctx context.Context, args []string, stdin io.Reader, stdout, stderr io. err = runDeliveryCommand(ctx, "retry", http.MethodPost, args[1:], stdout, stderr) case "cancel": err = runDeliveryCommand(ctx, "cancel", http.MethodPost, args[1:], stdout, stderr) + case "version": + fmt.Fprintf(stdout, "spoolctl %s\n", buildinfo.String()) + return 0 case "help", "-h", "--help": printUsage(stdout) return 0 @@ -388,6 +392,7 @@ func printUsage(output io.Writer) { fmt.Fprintln(output, " get inspect one delivery") fmt.Fprintln(output, " retry start a new retry cycle for a failed delivery") fmt.Fprintln(output, " cancel cancel pending or in-flight delivery work") + fmt.Fprintln(output, " version print version information") fmt.Fprintln(output) fmt.Fprintln(output, "Set SPOOLD_URL to override the default http://127.0.0.1:8080 API.") } diff --git a/internal/spoolctl/spoolctl_test.go b/internal/spoolctl/spoolctl_test.go index 4acee40..da9d4ba 100644 --- a/internal/spoolctl/spoolctl_test.go +++ b/internal/spoolctl/spoolctl_test.go @@ -304,6 +304,7 @@ func TestCommandUsageExitCodes(t *testing.T) { }{ {name: "no command", code: 2, text: "Usage:"}, {name: "help", args: []string{"help"}, code: 0, text: "crash-safe HTTP deliveries"}, + {name: "version", args: []string{"version"}, code: 0, text: "spoolctl dev"}, {name: "unknown", args: []string{"unknown"}, code: 2, text: "unknown command"}, {name: "send missing URL", args: []string{"send"}, code: 2, text: "requires exactly one target URL"}, {name: "list extra argument", args: []string{"list", "extra"}, code: 2, text: "does not accept"}, diff --git a/scripts/install.sh b/scripts/install.sh new file mode 100755 index 0000000..39e30bd --- /dev/null +++ b/scripts/install.sh @@ -0,0 +1,80 @@ +#!/bin/sh + +set -eu + +repository="Rionlyu/spoold" +install_dir="${SPOOLD_INSTALL_DIR:-/usr/local/bin}" +version="${SPOOLD_VERSION:-}" + +command -v curl >/dev/null 2>&1 || { + echo "spoold installer: curl is required" >&2 + exit 1 +} + +case "$(uname -s)" in + Linux) os="linux" ;; + Darwin) os="darwin" ;; + *) + echo "spoold installer: only Linux and macOS are supported" >&2 + exit 1 + ;; +esac + +case "$(uname -m)" in + x86_64 | amd64) arch="amd64" ;; + arm64 | aarch64) arch="arm64" ;; + *) + echo "spoold installer: unsupported architecture $(uname -m)" >&2 + exit 1 + ;; +esac + +if [ -z "$version" ]; then + release_url="$(curl -fsSL -o /dev/null -w '%{url_effective}' "https://github.com/${repository}/releases/latest")" + version="${release_url##*/}" +fi +case "$version" in + v*) ;; + *) version="v${version}" ;; +esac + +release_version="${version#v}" +asset="spoold_${release_version}_${os}_${arch}.tar.gz" +base_url="${SPOOLD_RELEASE_BASE_URL:-https://github.com/${repository}/releases/download/${version}}" +temporary="$(mktemp -d)" +trap 'rm -rf "$temporary"' EXIT HUP INT TERM + +curl -fsSL "${base_url}/${asset}" -o "${temporary}/${asset}" +curl -fsSL "${base_url}/checksums.txt" -o "${temporary}/checksums.txt" + +checksum_line="$( + awk -v asset="$asset" '$2 == asset { print }' "${temporary}/checksums.txt" +)" +if [ -z "$checksum_line" ] || [ "$(printf '%s\n' "$checksum_line" | wc -l | tr -d ' ')" -ne 1 ]; then + echo "spoold installer: release checksum for ${asset} is missing or ambiguous" >&2 + exit 1 +fi + +if command -v sha256sum >/dev/null 2>&1; then + ( + cd "$temporary" + printf '%s\n' "$checksum_line" | sha256sum --check - + ) +else + ( + cd "$temporary" + printf '%s\n' "$checksum_line" | shasum -a 256 --check - + ) +fi + +tar -xzf "${temporary}/${asset}" -C "$temporary" +archive_dir="${temporary}/spoold_${release_version}_${os}_${arch}" + +mkdir -p "$install_dir" +install -m 0755 "${archive_dir}/spoold" "${install_dir}/spoold" +install -m 0755 "${archive_dir}/spoolctl" "${install_dir}/spoolctl" + +"${install_dir}/spoold" -version >/dev/null +"${install_dir}/spoolctl" version >/dev/null + +echo "installed spoold and spoolctl ${version} to ${install_dir}" From d619bde42a5edb5938255495bad4232cad564bb3 Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 12:13:57 +0200 Subject: [PATCH 8/9] docs: prepare the public alpha launch --- .github/ISSUE_TEMPLATE/bug_report.yml | 44 ++++++ .github/ISSUE_TEMPLATE/config.yml | 5 + .github/ISSUE_TEMPLATE/feature_request.yml | 27 ++++ .github/dependabot.yml | 10 ++ .github/pull_request_template.md | 16 +++ CODE_OF_CONDUCT.md | 12 ++ CONTRIBUTING.md | 20 +++ README.md | 152 +++++++++++++++++---- SECURITY.md | 8 +- docs/design.md | 74 +++++++--- docs/launch-readiness.md | 143 +++++++++++++++++++ docs/product-direction.md | 45 +++--- 12 files changed, 491 insertions(+), 65 deletions(-) create mode 100644 .github/ISSUE_TEMPLATE/bug_report.yml create mode 100644 .github/ISSUE_TEMPLATE/config.yml create mode 100644 .github/ISSUE_TEMPLATE/feature_request.yml create mode 100644 .github/dependabot.yml create mode 100644 .github/pull_request_template.md create mode 100644 CODE_OF_CONDUCT.md create mode 100644 CONTRIBUTING.md create mode 100644 docs/launch-readiness.md diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml new file mode 100644 index 0000000..fdac895 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/bug_report.yml @@ -0,0 +1,44 @@ +name: Bug report +description: Report incorrect delivery, persistence, installation, or operational behavior +title: "bug: " +labels: + - bug +body: + - type: markdown + attributes: + value: Do not include secrets, authorization headers, or sensitive payloads. + - type: input + id: version + attributes: + label: Version + description: Output from `spoold -version` and `spoolctl version`. + validations: + required: true + - type: input + id: platform + attributes: + label: Platform + description: Operating system, architecture, and installation method. + validations: + required: true + - type: textarea + id: behavior + attributes: + label: What happened? + description: Include expected behavior and a minimal reproduction. + validations: + required: true + - type: textarea + id: logs + attributes: + label: Relevant sanitized logs + render: shell + - type: checkboxes + id: checks + attributes: + label: Checks + options: + - label: I removed secrets and sensitive payload data. + required: true + - label: I searched existing issues for this problem. + required: true diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml new file mode 100644 index 0000000..9e61ba3 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/config.yml @@ -0,0 +1,5 @@ +blank_issues_enabled: false +contact_links: + - name: Security vulnerability + url: https://github.com/Rionlyu/spoold/security/advisories/new + about: Report vulnerabilities privately diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml new file mode 100644 index 0000000..981cf40 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/feature_request.yml @@ -0,0 +1,27 @@ +name: Feature request +description: Propose an improvement to the focused local HTTP spool +title: "proposal: " +labels: + - enhancement +body: + - type: textarea + id: problem + attributes: + label: Problem + description: What real workflow or failure mode is not handled today? + validations: + required: true + - type: textarea + id: proposal + attributes: + label: Proposed behavior + description: Describe the smallest useful contract, including failure semantics. + validations: + required: true + - type: textarea + id: alternatives + attributes: + label: Alternatives considered + description: Explain why a script, existing HTTP client, or external queue is insufficient. + validations: + required: true diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..632e8eb --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,10 @@ +version: 2 +updates: + - package-ecosystem: github-actions + directory: / + schedule: + interval: weekly + - package-ecosystem: gomod + directory: / + schedule: + interval: weekly diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..fd0ef5e --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,16 @@ +## What changed + +Describe the user-visible behavior or failure mode this addresses. + +## Verification + +List the checks and failure scenarios you exercised. + +## Durability checklist + +- [ ] I ran `make verify`. +- [ ] Persistence or state-transition changes include replay coverage. +- [ ] Crash-sensitive changes include a process-level test where practical. +- [ ] Documentation describes any changed delivery or retention semantics. +- [ ] This change does not introduce secrets or sensitive payloads into tests, + logs, or fixtures. diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md new file mode 100644 index 0000000..c461ab1 --- /dev/null +++ b/CODE_OF_CONDUCT.md @@ -0,0 +1,12 @@ +# Code of Conduct + +Be respectful, constructive, and specific. Harassment, personal attacks, +discrimination, and disclosure of another person's private information are not +acceptable. + +Project maintainers may edit or remove contributions that violate these +expectations and may temporarily or permanently restrict participation. Report +conduct concerns privately to the repository owner through GitHub. + +This policy applies in project spaces and when representing the project in +public. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..b9e2ac0 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,20 @@ +# Contributing + +Bug reports, failure reproductions, portability fixes, and real deployment +notes are especially valuable while `spoold` is in alpha. + +Before opening a pull request: + +```sh +make verify +``` + +Changes to persistence or delivery state must include a replay or transition +test. Changes to crash behavior should include a process-level test when +possible. Keep the project focused on one-machine durable HTTP delivery; a +dashboard, hosted control plane, and distributed coordination are deliberate +non-goals. + +Use GitHub's private security-advisory feature for vulnerabilities rather than +a public issue. By participating, you agree to follow the +[Code of Conduct](CODE_OF_CONDUCT.md). diff --git a/README.md b/README.md index cade79f..6939dfb 100644 --- a/README.md +++ b/README.md @@ -21,13 +21,65 @@ account, or a language-specific SDK: - a strict JSON API, Prometheus-format metrics, and structured logs; - no runtime dependencies outside the Go standard library. -## Quick start +Use it when a cron job must report completion after a reboot, an edge agent +must survive hours offline, an appliance must retain callbacks without shipping +a database, or a small service needs a local outbound reliability boundary. +Unlike `curl --retry`, the retry lifecycle is independent of the submitting +process. Do not use it when enqueueing must be atomic with an application +database transaction or when the queue must be highly available across hosts. + +## Install + +Release archives support Linux and macOS on amd64 and arm64. Every archive +contains both `spoold` and `spoolctl`, plus the license and security policy. + +```sh +curl -fsSL https://raw.githubusercontent.com/Rionlyu/spoold/main/scripts/install.sh | sh +``` + +Set `SPOOLD_INSTALL_DIR` to install somewhere other than `/usr/local/bin`, or +set `SPOOLD_VERSION` to install a specific tag. The installer verifies the +archive against the release checksum. Release checksums are keylessly signed +with Sigstore, and the release workflow also publishes SBOMs and GitHub build +provenance. + +To authenticate a downloaded checksum file with Cosign: + +```sh +tag=v0.1.0 +cosign verify-blob checksums.txt \ + --bundle checksums.txt.sigstore.json \ + --certificate-identity \ + "https://github.com/Rionlyu/spoold/.github/workflows/release.yml@refs/tags/${tag}" \ + --certificate-oidc-issuer "https://token.actions.githubusercontent.com" +``` + +Tagged multi-architecture images are published to +`ghcr.io/rionlyu/spoold`. The container must have a persistent volume mounted at +`/var/lib/spoold`. + +```sh +docker run -d \ + --name spoold \ + --restart unless-stopped \ + --publish 127.0.0.1:8080:8080 \ + --volume spoold-data:/var/lib/spoold \ + ghcr.io/rionlyu/spoold:v0.1.0 +``` -Go 1.26 or newer is required. +To build from source instead, Go 1.26 or newer is required: ```sh make build -./bin/spoold -allow-private-targets +``` + +Source-built binaries are written to `./bin`; prefix the following `spoold` and +`spoolctl` commands with `./bin/` when they are not installed on `PATH`. + +## Quick start + +```sh +spoold -allow-private-targets ``` Private targets are disabled by default. The flag is needed for the local @@ -40,7 +92,7 @@ go run ./examples/receiver Enqueue a delivery: ```sh -./bin/spoolctl send \ +spoolctl send \ --idempotency-key build-2026-07-16 \ --header 'X-Event-Type: build.completed' \ --data '{"repository":"spoold","result":"passed"}' \ @@ -56,9 +108,9 @@ idempotency conflict. Inspect the delivery and metrics: ```sh -./bin/spoolctl list -./bin/spoolctl get -./bin/spoolctl retry +spoolctl list +spoolctl get +spoolctl retry curl http://127.0.0.1:8080/metrics ``` @@ -67,6 +119,17 @@ Use `--json` for machine-readable output and `spoolctl help` for the complete command list. The HTTP API remains the stable integration surface for programs that do not use the CLI. +For an owner-only local endpoint instead of TCP: + +```sh +spoold \ + -unix-socket "$HOME/.spoold/spoold.sock" \ + -journal "$HOME/.spoold/spoold.journal" + +export SPOOLD_URL="unix://$HOME/.spoold/spoold.sock" +spoolctl list +``` + ## API | Method | Path | Purpose | @@ -76,12 +139,18 @@ that do not use the CLI. | `GET` | `/v1/deliveries/{id}` | Inspect one delivery | | `POST` | `/v1/deliveries/{id}/cancel` | Cancel pending or leased work | | `POST` | `/v1/deliveries/{id}/retry` | Start a new retry cycle for failed work | -| `GET` | `/healthz`, `/readyz` | Process health | +| `GET` | `/healthz`, `/readyz` | Liveness and persistence readiness | | `GET` | `/metrics` | Prometheus text exposition | -Request bodies are limited to 1 MiB and reject unknown fields. Delivery bodies -must contain valid JSON. The method defaults to `POST`, and `maxAttempts` -defaults to `8`. +Requests reject unknown fields. A delivery body can use `body` for JSON or +`bodyBase64` for arbitrary bytes; the fields are mutually exclusive and the +decoded body is limited to 1 MiB. `spoolctl send --data-binary ` handles +the base64 transport automatically. The method defaults to `POST`, and +`maxAttempts` defaults to `8`. + +List responses omit headers and bodies to remain bounded and avoid exposing +secrets in bulk output. `GET /v1/deliveries/{id}` returns the complete retained +request. Workers treat transport errors, `408`, `425`, `429`, and `5xx` responses as retryable. Other non-success responses fail immediately. Backoff starts at one @@ -113,13 +182,19 @@ during shutdown, ignores an incomplete final append, and rejects corruption in any earlier complete record. Online compaction replaces superseded records with one current record per -delivery. It is lossless: pending, in-flight, succeeded, failed, and canceled -deliveries remain present, as do request fingerprints used for idempotency. -Compaction briefly pauses store mutations while it writes and synchronizes a -snapshot, then atomically replaces the journal. The default background check -runs once per minute and compacts only when the journal is at least 64 MiB and -contains at least twice as many physical records as live deliveries. A -compaction failure is logged and counted but does not stop delivery processing. +retained delivery. Compaction itself is lossless and briefly pauses store +mutations while it writes and synchronizes a snapshot, then atomically replaces +the journal. Separately, terminal retention defaults to seven days. Expired +succeeded, failed, and canceled deliveries are tombstoned, their idempotency +keys expire with them, and the journal is compacted. Pending and in-flight work +is never removed by retention. + +The default background check runs once per minute and also compacts when the +journal is at least 64 MiB and contains at least twice as many physical records +as retained deliveries. A maintenance failure is logged and counted but does +not terminate the process. If replacement was renamed but its directory could +not be synchronized, readiness fails and mutations pause until the next enabled +maintenance pass repairs the journal. See [the design document](docs/design.md) for the state machine and durability boundary and replacement sequence. @@ -135,6 +210,10 @@ Use `-allow-private-targets` only in a trusted development environment. `spoold` does not provide authentication or tenant isolation in v0.1 and should not be exposed directly to an untrusted network. +Only one process can own a journal. A second process fails startup rather than +risk interleaved writes. On Linux and macOS, `-unix-socket` creates an +owner-only `0600` socket and is the preferred boundary on a multi-user machine. + ## Configuration ```text @@ -142,8 +221,17 @@ exposed directly to an untrusted network. HTTP listen address (default "127.0.0.1:8080") -journal string append-only journal path (default "data/spoold.journal") +-max-journal-bytes int + reject new deliveries at this physical journal size; 0 disables + (default 1073741824) +-terminal-retention duration + retain terminal delivery history; 0 retains forever (default 168h) -workers int number of delivery workers (default 4) +-per-target-workers int + maximum concurrent requests to one target origin (default 1) +-unix-socket string + owner-only Unix socket path; overrides -listen -allow-private-targets allow private and loopback delivery targets -request-timeout duration @@ -151,13 +239,15 @@ exposed directly to an untrusted network. -shutdown-timeout duration graceful shutdown timeout (default 10s) -compact-threshold-bytes int - minimum journal size for compaction; 0 disables (default 67108864) + minimum journal size for redundancy compaction; 0 disables the size + trigger (default 67108864) -compact-check-interval duration - journal compaction check interval (default 1m) + journal maintenance interval (default 1m) ``` -Container builds listen on `0.0.0.0:8080` and store the journal under -`/var/lib/spoold`. +Container builds listen on `0.0.0.0:8080`, store the journal under +`/var/lib/spoold`, and enforce the same 1 GiB admission limit and seven-day +terminal retention defaults. ## Development @@ -168,6 +258,9 @@ make test make test-race make build make verify + +# Exercise a real daemon kill, journal reopen, and recovered delivery. +go test ./cmd/spoold -run TestProcessRecoversDeliveryAfterSIGKILL -v ``` ## Project structure @@ -176,6 +269,7 @@ make verify cmd/spoold/ process configuration and graceful shutdown cmd/spoolctl/ curl-like local delivery client internal/api/ strict HTTP API and metrics exposition +internal/buildinfo/ release version, commit, and build metadata internal/compactor/ background journal compaction policy internal/delivery/ delivery model and request fingerprinting internal/spoolctl/ client commands, API transport, and output @@ -183,22 +277,26 @@ internal/store/ append-only journal and state transitions internal/target/ SSRF-aware HTTP transport internal/worker/ leases, delivery, retry policy, and counters examples/receiver/ local HTTP destination for the quick start +scripts/install.sh checksummed release installer docs/design.md semantics and deliberate non-goals +docs/launch-readiness.md public release gate and positioning brief ``` The focused product decision and competitive boundary are documented in -[product direction](docs/product-direction.md). +[product direction](docs/product-direction.md). The pre-release checks and +market boundary are in [public launch readiness](docs/launch-readiness.md). ## Limitations -- One process owns one journal file. -- Compaction retains terminal deliveries; there is no retention-based purge or - administrative compaction endpoint. +- One process owns one journal file; concurrent ownership is rejected. +- Terminal history and its idempotency keys expire after seven days by default. + There is no administrative purge endpoint. - There is no authentication, tenant isolation, rate limiting, or per-target - concurrency control. + rate shaping. - Response bodies are retained only as a bounded error string, not as durable artifacts. - `spoold` does not claim distributed coordination or exactly-once delivery. +- Release binaries currently target Linux and macOS. ## License diff --git a/SECURITY.md b/SECURITY.md index 4a0299a..cc5888a 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -8,10 +8,14 @@ feature rather than opening a public issue. The supported security boundary is the default single-user deployment: - the API listens on loopback unless configured otherwise; +- on Linux and macOS, an owner-only Unix socket is available for stronger local + isolation; - private and loopback delivery targets are rejected unless explicitly allowed; - the journal is created with owner-only file permissions; +- concurrent ownership of one journal is rejected; - outbound proxy environment variables are ignored. `spoold` v0.1 has no authentication or tenant isolation. Exposing its API to an -untrusted network is unsupported. - +untrusted network is unsupported. Treat delivery headers and bodies as +sensitive because they are retained in the journal until terminal retention +expires. diff --git a/docs/design.md b/docs/design.md index 30dbc70..2ad3764 100644 --- a/docs/design.md +++ b/docs/design.md @@ -38,20 +38,47 @@ destination cannot be retracted. The journal is newline-delimited JSON. Each mutation appends the complete current delivery record under a versioned envelope and calls `fsync` before -returning. Replay keeps the last valid record for each delivery. +returning. Creation of a new journal also synchronizes its directory before the +first enqueue can be acknowledged. Replay keeps the last valid record for each +delivery. A final partial line is treated as an interrupted append and ignored. Corrupt complete records fail startup rather than silently discarding acknowledged state. +One process owns a journal through a stable adjacent lock file. A competing +process fails startup instead of interleaving writes. An append or sync failure +makes persistence readiness sticky: new mutations fail until a successful +compaction establishes and synchronizes a replacement journal. The next enabled +maintenance pass attempts that repair even below the size threshold. Liveness +remains separate so an operator can distinguish a running process from a +durable one. + +Journal format version 2 can record either a complete delivery or a deletion +tombstone. Replay remains compatible with version 1 records. + +### Capacity and retention + +The physical journal has a configurable admission limit, 1 GiB by default. +Crossing it rejects only new deliveries; idempotent lookup and transitions for +already accepted work remain available. This keeps existing work recoverable +without allowing unbounded new commitments. + +Terminal retention defaults to seven days. Maintenance writes deletion +tombstones for expired succeeded, failed, and canceled deliveries, then +compacts them away. Their idempotency keys expire at the same time. Pending and +in-flight deliveries are never removed by retention. Setting retention to zero +keeps terminal deliveries indefinitely. + ### Online compaction Compaction is a lossless replacement of superseded physical records, not a -retention policy. The store mutex remains held for the snapshot, so creates, -claims, completions, retries, and cancellations pause briefly. Every live -delivery is written exactly once in delivery-ID order, including pending, -in-flight, succeeded, failed, and canceled deliveries. The associated request -fingerprint is carried forward so idempotency behavior is unchanged. +retention policy: retention decides which terminal records are still live +before compaction begins. The store mutex remains held for the snapshot, so +creates, claims, completions, retries, and cancellations pause briefly. Every +retained delivery is written exactly once in delivery-ID order, including +pending, in-flight, succeeded, failed, and canceled deliveries. The associated +request fingerprint is carried forward so idempotency behavior is unchanged. The atomic replacement sequence is: @@ -64,17 +91,20 @@ The atomic replacement sequence is: descriptor. If compaction fails before the rename, the original journal remains active and -the temporary file is removed. If an error occurs after the rename, the store -adopts the replacement descriptor before returning the error, so subsequent -appends remain recoverable. Abandoned temporary compaction files are removed -when the store opens. Failures are counted and logged but are non-fatal to the -service. - -The background compactor checks once per minute by default. It runs only when -the journal is at least 64 MiB and its physical record count is at least twice -the live-delivery count. The byte threshold and check interval are configurable, -and a zero byte threshold disables automatic compaction. There is deliberately -no administrative HTTP endpoint and no purge operation. +the temporary file is removed. If an error occurs after the rename but before +directory synchronization, the store adopts the replacement descriptor but +marks persistence unready and blocks mutations until a later compaction repairs +the durability boundary. An error after directory synchronization still adopts +the durable replacement and subsequent appends remain recoverable. Abandoned +temporary compaction files are removed when the store opens. Failures are +counted and logged rather than terminating the process. + +The background compactor checks once per minute by default. Size-based +compaction runs only when the journal is at least 64 MiB and its physical record +count is at least twice the live-delivery count. Retention and persistence +repair are evaluated independently. The byte threshold and check interval are +configurable, and a zero byte threshold disables size-based compaction. There +is deliberately no administrative HTTP endpoint and no purge operation. ## Concurrency @@ -84,6 +114,11 @@ updates include the attempt number; a late worker cannot overwrite a newer claim. Compaction uses the same serialization boundary, trading a brief mutation pause for a snapshot that cannot omit or reorder a concurrent state change. +Workers also reserve target origins while claiming work. The default allows one +active request per scheme/host/port while still using the global worker pool, +so an unavailable destination cannot occupy every worker. The limit is +configurable. + ## Networking Only HTTP and HTTPS targets are accepted. User information and URL fragments @@ -91,10 +126,13 @@ are rejected. The production-default dialer rejects loopback, private, link-local, unspecified, and multicast addresses after DNS resolution. Local targets can be enabled explicitly for development and integration tests. +The control API listens on loopback by default. On Linux and macOS it can +instead use an owner-only Unix socket, which is the preferred boundary on a +multi-user host. ## Non-goals for v0.1 -- Multi-process access to one journal. +- Sharing one journal between processes. - Distributed worker coordination. - Exactly-once delivery. - Arbitrary scheduling or recurring jobs. diff --git a/docs/launch-readiness.md b/docs/launch-readiness.md new file mode 100644 index 0000000..065b087 --- /dev/null +++ b/docs/launch-readiness.md @@ -0,0 +1,143 @@ +# Public launch readiness + +This document is the release gate and positioning brief for the first public +`spoold` release. It intentionally separates work required to make the +durability promise credible from features that would dilute the product. + +## Launch decision + +Launch `spoold` as **durable curl**: the smallest self-hosted component that +accepts an arbitrary outbound HTTP request only after synchronizing it to local +disk, then keeps delivering it across destination outages and process +restarts. + +The promise is deliberately narrow: + +> Accept the request now, deliver it later, and require no database, broker, or +> cloud service. + +Do not position it as a webhook platform, message broker, transactional outbox, +or exactly-once system. Those products solve larger or different problems. + +## Market boundary + +The adjacent products validate the underlying store-and-forward need but make +different tradeoffs: + +| Category | Examples | Why `spoold` remains distinct | +|---|---|---| +| General HTTP forwarders | [aswh](https://github.com/pepmartinez/aswh) | Similar arbitrary-HTTP goal, but its supported backends are MongoDB, PostgreSQL, and Redis. | +| Self-hosted webhook platforms | [Hookdeck Outpost](https://github.com/hookdeck/outpost), [Convoy](https://github.com/frain-dev/convoy) | Broader routing and platform scope with additional runtime infrastructure. | +| Hosted delivery queues | [Svix](https://docs.svix.com/retries), [QStash](https://upstash.com/docs/qstash/features/retry) | The provider owns the queue; `spoold` is local, offline-capable, and account-free. | +| Telemetry agents | [Vector](https://vector.dev/docs/reference/configuration/sinks/http/), [Fluent Bit](https://docs.fluentbit.io/manual/data-pipeline/buffering) | Disk buffering serves a telemetry pipeline rather than an explicit fsync-backed API for caller-selected HTTP requests. | +| Transactional outboxes | Framework- and database-specific implementations | They can be atomic with application state; `spoold` begins its guarantee only after a separate enqueue reaches the daemon. | + +The defensible claim is not that HTTP retries are new. It is that this +combination is unusually small: arbitrary HTTP, synchronization before +acknowledgement, deterministic local replay, and no runtime service dependency. + +## Technical release gates + +The first release must preserve every checked property below: + +- [x] A successful enqueue means the journal append has been synchronized. +- [x] Replay rejects complete-record corruption and tolerates only an + interrupted final append. +- [x] A process killed during an outage can restart from the same journal and + complete the accepted delivery. +- [x] A second process cannot concurrently own the same journal. +- [x] Storage failure makes readiness fail and blocks further mutations until + a synchronized replacement journal repairs the persistence boundary. +- [x] New admission is bounded while transitions for accepted work remain + available. +- [x] Pending and in-flight work is never removed by retention. +- [x] Idempotency survives restart and compaction, then expires with retained + terminal history. +- [x] One unavailable target cannot occupy every worker. +- [x] Arbitrary byte bodies and JSON bodies are both round-tripped losslessly. +- [x] The default network policy blocks private target classes and redirects + cannot bypass it. +- [x] Loopback and owner-only Unix socket control-plane boundaries are + available. +- [x] Binaries embed version metadata and ship together in checksummed + Linux/macOS archives for amd64 and arm64. +- [x] The release workflow is configured to produce SBOMs, keyless signatures, + provenance, and a non-root multi-architecture container. + +These gates are exercised by unit, race, process, packaging, and container +checks in CI. A tag is publishable only from a green `main`. + +## Known alpha constraints + +These are disclosure requirements, not launch blockers: + +- delivery is at-least-once; a destination must deduplicate by the stable + delivery ID when duplicate effects matter; +- `spoold` cannot atomically join an application's database transaction; +- one process and one host own a journal; there is no high availability; +- the control API has no network authentication and must remain on loopback, an + owner-only Unix socket, or another trusted boundary; +- headers and bodies are stored in the journal and should be protected as + sensitive data; +- terminal history and its idempotency keys expire after seven days by default; +- request bodies are limited to 1 MiB, response bodies are not archived, and + Windows is not a release target; +- durability ultimately depends on the host filesystem and storage honoring + synchronization calls. + +Authentication, a dashboard, distributed workers, transformations, scheduling, +fan-out, and a hosted control plane should not delay the first release. + +## Release procedure + +1. Merge only after both CI jobs pass on the launch pull request. +2. Confirm `main` CI is green and the release notes describe at-least-once + behavior, the trust boundary, retention, and platform support. +3. Tag the immutable commit as `v0.1.0` and let the release workflow publish + archives, checksums, signatures, SBOMs, provenance, and the GHCR image. +4. On a clean supported host, install with `scripts/install.sh`, confirm + `spoold -version` and `spoolctl version`, enqueue while the destination is + unavailable, terminate the daemon, restart it, and observe successful + delivery. +5. Verify the checksum signature and GitHub attestations, pull both container + architectures, and confirm the container writes successfully to a mounted + persistent volume and fails clearly when the journal location is read-only. +6. Publish the announcement only after the released artifacts—not local + builds—pass the smoke test. + +Never move or overwrite a published tag. If the release is defective, document +the defect and publish a new patch version. + +## Launch message and demo + +The launch should lead with the failure it removes: + +> `curl` tells you whether a request worked now. `spoold` tells you the request +> is safely queued and keeps trying after your script, destination, or machine +> comes back. + +The proof should be a short terminal recording, not a feature tour: + +1. start `spoold` while the example destination is offline; +2. enqueue one request and show `queued`; +3. kill the daemon with `SIGKILL`; +4. start the destination and restart `spoold`; +5. show the same delivery reach `succeeded`. + +Link directly to the durability design and the process-level test so technical +readers can verify the promise. + +## Post-launch validation + +The goal is useful adoption, not raw impressions. During the first 30 days, +seek: + +- three unrelated users who reproduce the crash-recovery demo; +- one real use in a script, edge agent, appliance, or small self-hosted service; +- one report from a prolonged destination outage or forced reboot; +- installation feedback from every supported OS and architecture; +- concrete reasons from users who considered `spoold` but chose another tool. + +Prioritize defects that threaten acknowledged work, installation failures, and +confusing durability semantics. Add features only when repeated real workflows +show that the focused contract is insufficient. diff --git a/docs/product-direction.md b/docs/product-direction.md index b04032d..dea9026 100644 --- a/docs/product-direction.md +++ b/docs/product-direction.md @@ -34,12 +34,14 @@ It provides a clear durability boundary once the caller submits the delivery. The underlying store-and-forward pattern is established. The product differentiation is the combination of: -- arbitrary caller-selected HTTP requests rather than one telemetry schema; +- arbitrary caller-selected HTTP requests and byte bodies rather than one + telemetry schema; - synchronization before enqueue acknowledgement; - deterministic crash replay and lease recovery; - idempotent submission with conflicting-reuse detection; - safe outbound target resolution; -- one self-contained daemon and one local journal; +- one self-contained daemon, one exclusively owned local journal, and an + optional owner-only Unix socket; - no PostgreSQL, MongoDB, Redis, broker, cloud account, or language SDK. Current adjacent categories make different tradeoffs: @@ -62,22 +64,25 @@ The direction decision was re-evaluated on 2026-07-18 with an estimated 92% confidence. This is a product judgment, not a statistical measurement. The research included: -- [aswh](https://codeberg.org/pepmartinez/aswh), the closest general +- [aswh](https://github.com/pepmartinez/aswh), the closest general store-and-forward HTTP proxy, which requires MongoDB, Redis, or PostgreSQL; +- [Hookdeck Outpost](https://github.com/hookdeck/outpost) and + [Convoy](https://github.com/frain-dev/convoy), self-hosted webhook delivery + systems with a broader feature and infrastructure surface; - [Svix](https://docs.svix.com/retries) and [QStash](https://upstash.com/docs/qstash/features/retry), which provide hosted delivery infrastructure; - [Vector](https://vector.dev/docs/reference/configuration/sinks/http/) and - [Fluent Bit](https://docs.fluentbit.io/manual/administration/buffering-and-storage), + [Fluent Bit](https://docs.fluentbit.io/manual/data-pipeline/buffering), which provide disk-buffered telemetry pipelines; - framework-specific transactional outbox implementations that share the application's database. -The direction remains worthy because no maintained alternative found in the -audit combined arbitrary HTTP submission, synchronization before -acknowledgement, single-host crash replay, and zero runtime infrastructure. -This is a differentiation claim, not a claim that store-and-forward HTTP has no -prior art. +The direction remains worthy because the audit did not find a maintained +alternative that combined arbitrary HTTP submission, an explicit +synchronization-before-acknowledgement contract, single-host crash replay, and +zero runtime infrastructure. This is a narrow differentiation claim based on +the researched set, not a claim that store-and-forward HTTP has no prior art. Development should stop or reposition if a maintained, dependency-free local HTTP spool with comparable durability becomes clearly established and `spoold` @@ -87,17 +92,21 @@ distribution and feedback work. ## Product contract -The durable local-spool direction requires: +The public-alpha contract is: 1. a curl-like client suitable for humans and scripts; -2. a local trust boundary that does not require a public API; -3. bounded retention and explicit disk-full behavior; -4. per-destination backpressure so one target cannot monopolize delivery; -5. packaging as signed binaries and a minimal container; -6. failure tests covering abrupt termination, restart, and storage faults. - -The first product milestone adds `spoolctl`, a dependency-free client for -submitting, inspecting, retrying, and canceling deliveries. +2. JSON and arbitrary-byte request bodies with stable idempotency fingerprints; +3. a local trust boundary through loopback or an owner-only Unix socket; +4. exclusive journal ownership, bounded retention, readiness reporting, and + explicit disk-full behavior; +5. per-target concurrency so one unavailable destination cannot monopolize + delivery; +6. checksummed and signed binaries, SBOMs, provenance, and a minimal container; +7. process-level failure tests covering abrupt termination, restart, replay, + and storage faults. + +These requirements are implemented. The next milestone is deliberately an +alpha release and external use, not another expansion of the feature set. ## Deliberate non-goals From 177e76180f03647013c8f43cedfc8064c38b7c69 Mon Sep 17 00:00:00 2001 From: Rionlyu Date: Sat, 18 Jul 2026 12:17:05 +0200 Subject: [PATCH 9/9] release: use supported prerelease template --- .goreleaser.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.goreleaser.yml b/.goreleaser.yml index 98a9157..6c916d4 100644 --- a/.goreleaser.yml +++ b/.goreleaser.yml @@ -98,7 +98,7 @@ dockers_v2: - ghcr.io/rionlyu/spoold tags: - "{{ .Tag }}" - - "{{ if not .IsPrerelease }}latest{{ end }}" + - "{{ if not .Prerelease }}latest{{ end }}" dockerfile: .goreleaser.Dockerfile platforms: - linux/amd64