release: 0.1.0: Phases 0-11: panel, Postfix/DKIM, milter, backup, Docker image
Initial feature-complete implementation of the v1.0 specification (phases 0-11 of `docs/implementation-plan.md`). Added: - Panel (Go, single static binary) with SQLite persistence, one-time - Domain management with per-domain DKIM (RSA-2048, generated in pure Go) and - Application (sender identity) management: SASL credentials via `sasldb2`, - Full Postfix relay config generated from env at container start: SMTPS 465, - Journal milter (pure Go, `go-milter`) recording every send to `send_log`; … (6 more in CHANGELOG) Full detail: CHANGELOG.md (0.1.0).
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
// Package logtail follows Postfix's mail.log and reconciles the send-log
|
||||
// delivery statuses the journal-milter could not know at receive time (spec
|
||||
// 7.3). A milter row starts life as "queued"; Postfix only decides sent /
|
||||
// deferred / bounced later, per recipient, and reports it in mail.log. This
|
||||
// package parses those lines by queue-id + recipient and advances the matching
|
||||
// rows, and prunes rows past the retention window.
|
||||
package logtail
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"codeberg.org/mix/selfpost/internal/store"
|
||||
)
|
||||
|
||||
// StatusStore is the slice of the store the log-tailer needs. *store.Store
|
||||
// satisfies it.
|
||||
type StatusStore interface {
|
||||
UpdateStatus(queueID, recipient, status string) (int64, error)
|
||||
DeleteSendLogBefore(cutoff time.Time) (int64, error)
|
||||
}
|
||||
|
||||
// pollInterval is how often the tail loop checks for new bytes / rotation. It
|
||||
// is a var so tests can shorten it.
|
||||
var pollInterval = time.Second
|
||||
|
||||
const (
|
||||
// retentionInterval is how often the retention sweep runs (also once at
|
||||
// startup). The window itself is configurable; the cadence need not be.
|
||||
retentionInterval = 6 * time.Hour
|
||||
// defaultRetentionDays applies when the configured value is unset/invalid
|
||||
// (spec 7.3).
|
||||
defaultRetentionDays = 90
|
||||
)
|
||||
|
||||
// deliveryRe matches a Postfix delivery line and captures queue-id, recipient
|
||||
// and status, e.g.
|
||||
//
|
||||
// postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=…, dsn=2.0.0, status=sent (250 OK)
|
||||
//
|
||||
// The "<queue-id>: to=<addr>, …, status=<word>" shape is specific to the
|
||||
// delivery agents; qmgr/smtpd/cleanup lines do not match.
|
||||
var deliveryRe = regexp.MustCompile(`\b([0-9A-Za-z]+): to=<([^>]*)>,.*\bstatus=(\w+)`)
|
||||
|
||||
// parseDelivery extracts (queue-id, recipient, status) from a mail.log line.
|
||||
// ok is false for lines that are not recognised delivery results.
|
||||
func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
|
||||
m := deliveryRe.FindStringSubmatch(line)
|
||||
if m == nil {
|
||||
return "", "", "", false
|
||||
}
|
||||
switch m[3] {
|
||||
case "sent":
|
||||
status = store.StatusSent
|
||||
case "deferred":
|
||||
status = store.StatusDeferred
|
||||
case "bounced":
|
||||
status = store.StatusBounced
|
||||
case "expired":
|
||||
// Postfix gave up after the queue lifetime; a final failure for us.
|
||||
status = store.StatusBounced
|
||||
default:
|
||||
return "", "", "", false
|
||||
}
|
||||
return m[1], m[2], status, true
|
||||
}
|
||||
|
||||
// Run follows path and updates send-log statuses until ctx is cancelled, while
|
||||
// a background sweep prunes rows older than retentionDays. It returns nil on a
|
||||
// clean shutdown.
|
||||
func Run(ctx context.Context, path string, st StatusStore, retentionDays int) error {
|
||||
go retentionLoop(ctx, st, retentionDays)
|
||||
|
||||
return follow(ctx, path, func(line string) {
|
||||
queueID, recipient, status, ok := parseDelivery(line)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if _, err := st.UpdateStatus(queueID, recipient, status); err != nil {
|
||||
log.Printf("log-tailer: update %s/%s -> %s: %v", queueID, recipient, status, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// retentionLoop prunes expired send-log rows immediately and then periodically.
|
||||
func retentionLoop(ctx context.Context, st StatusStore, retentionDays int) {
|
||||
if retentionDays <= 0 {
|
||||
retentionDays = defaultRetentionDays
|
||||
}
|
||||
prune := func() {
|
||||
cutoff := time.Now().UTC().AddDate(0, 0, -retentionDays)
|
||||
n, err := st.DeleteSendLogBefore(cutoff)
|
||||
if err != nil {
|
||||
log.Printf("log-tailer: retention prune: %v", err)
|
||||
return
|
||||
}
|
||||
if n > 0 {
|
||||
log.Printf("log-tailer: pruned %d send-log rows older than %d days", n, retentionDays)
|
||||
}
|
||||
}
|
||||
|
||||
prune()
|
||||
t := time.NewTicker(retentionInterval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
prune()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TailLines returns up to n of the most recent lines from path, for the
|
||||
// panel's mail.log monitoring view (spec 7.2.13). It is a one-shot,
|
||||
// point-in-time read on request — unrelated to the background follow loop
|
||||
// above — that reads backwards in chunks so it stays cheap against a
|
||||
// multi-megabyte log rather than reading the whole file every poll.
|
||||
func TailLines(path string, n int) ([]string, error) {
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
const chunkSize = 8192
|
||||
var (
|
||||
buf []byte
|
||||
offset = info.Size()
|
||||
)
|
||||
for offset > 0 && bytes.Count(buf, []byte("\n")) <= n {
|
||||
size := int64(chunkSize)
|
||||
if size > offset {
|
||||
size = offset
|
||||
}
|
||||
offset -= size
|
||||
chunk := make([]byte, size)
|
||||
if _, err := f.ReadAt(chunk, offset); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
buf = append(chunk, buf...)
|
||||
}
|
||||
|
||||
text := strings.TrimRight(string(buf), "\n")
|
||||
if text == "" {
|
||||
return nil, nil
|
||||
}
|
||||
lines := strings.Split(text, "\n")
|
||||
if len(lines) > n {
|
||||
lines = lines[len(lines)-n:]
|
||||
}
|
||||
return lines, nil
|
||||
}
|
||||
|
||||
// follow tails path line by line, calling handle for each complete line, until
|
||||
// ctx is cancelled. It starts at end-of-file (so a restart does not reprocess
|
||||
// history) and reopens the file when it is rotated (inode change from
|
||||
// logrotate's create, or truncation from copytruncate) so nothing is missed.
|
||||
func follow(ctx context.Context, path string, handle func(string)) error {
|
||||
var (
|
||||
f *os.File
|
||||
r *bufio.Reader
|
||||
info os.FileInfo
|
||||
pending string
|
||||
)
|
||||
openAt := func(offset int64, whence int) error {
|
||||
nf, err := os.Open(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := nf.Seek(offset, whence); err != nil {
|
||||
nf.Close()
|
||||
return err
|
||||
}
|
||||
ni, err := nf.Stat()
|
||||
if err != nil {
|
||||
nf.Close()
|
||||
return err
|
||||
}
|
||||
if f != nil {
|
||||
f.Close()
|
||||
}
|
||||
f, r, info, pending = nf, bufio.NewReader(nf), ni, ""
|
||||
return nil
|
||||
}
|
||||
|
||||
// The container may start before Postfix has created mail.log; wait for it.
|
||||
for {
|
||||
if err := openAt(0, io.SeekEnd); err == nil {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-time.After(pollInterval):
|
||||
}
|
||||
}
|
||||
defer func() {
|
||||
if f != nil {
|
||||
f.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
drain := func() {
|
||||
for {
|
||||
line, err := r.ReadString('\n')
|
||||
if err == io.EOF {
|
||||
pending += line // hold the partial line until it completes
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("log-tailer: read %s: %v", path, err)
|
||||
return
|
||||
}
|
||||
full := pending + line
|
||||
pending = ""
|
||||
handle(strings.TrimRight(full, "\r\n"))
|
||||
}
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(pollInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-ticker.C:
|
||||
drain()
|
||||
ni, err := os.Stat(path)
|
||||
if err != nil {
|
||||
continue // file briefly gone mid-rotation; try again next tick
|
||||
}
|
||||
pos, _ := f.Seek(0, io.SeekCurrent)
|
||||
if !os.SameFile(info, ni) || ni.Size() < pos {
|
||||
// Rotated away or truncated: the old (renamed) inode may have
|
||||
// gained lines between the drain() above and this check, since
|
||||
// Postfix keeps writing to it until it reloads. Drain it once
|
||||
// more before switching so nothing in that gap is lost.
|
||||
drain()
|
||||
if err := openAt(0, io.SeekStart); err != nil {
|
||||
log.Printf("log-tailer: reopen %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
package logtail
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"codeberg.org/mix/selfpost/internal/store"
|
||||
)
|
||||
|
||||
func TestParseDelivery(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
line string
|
||||
wantOK bool
|
||||
queueID, recipient, status string
|
||||
}{
|
||||
{
|
||||
name: "sent",
|
||||
line: "2026-07-11T11:55:34 host postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=mx.example.net[203.0.113.9]:25, delay=0.5, dsn=2.0.0, status=sent (250 OK)",
|
||||
wantOK: true,
|
||||
queueID: "41E862C00D9E",
|
||||
recipient: "a@example.net",
|
||||
status: store.StatusSent,
|
||||
},
|
||||
{
|
||||
name: "deferred",
|
||||
line: "host postfix/smtp[26]: 5900C2C00D9E: to=<y@example.net>, relay=none, delay=30, dsn=4.4.1, status=deferred (connect timed out)",
|
||||
wantOK: true,
|
||||
queueID: "5900C2C00D9E",
|
||||
recipient: "y@example.net",
|
||||
status: store.StatusDeferred,
|
||||
},
|
||||
{
|
||||
name: "bounced",
|
||||
line: "host postfix/smtp[26]: ABC: to=<no@example.net>, relay=…, dsn=5.1.1, status=bounced (user unknown)",
|
||||
wantOK: true,
|
||||
queueID: "ABC",
|
||||
recipient: "no@example.net",
|
||||
status: store.StatusBounced,
|
||||
},
|
||||
{
|
||||
name: "expired maps to bounced",
|
||||
line: "host postfix/smtp[26]: DEF: to=<slow@example.net>, relay=none, status=expired (delivery temporarily suspended)",
|
||||
wantOK: true,
|
||||
queueID: "DEF",
|
||||
recipient: "slow@example.net",
|
||||
status: store.StatusBounced,
|
||||
},
|
||||
{
|
||||
name: "qmgr from-line ignored",
|
||||
line: "host postfix/qmgr[10]: 41E862C00D9E: from=<noreply@example.com>, size=500, nrcpt=1 (queue active)",
|
||||
wantOK: false,
|
||||
},
|
||||
{
|
||||
name: "smtpd client-line ignored",
|
||||
line: "host postfix/smtpd[10]: 41E862C00D9E: client=unknown[203.0.113.7]",
|
||||
wantOK: false,
|
||||
},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
q, r, s, ok := parseDelivery(c.line)
|
||||
if ok != c.wantOK {
|
||||
t.Fatalf("ok = %v, want %v", ok, c.wantOK)
|
||||
}
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if q != c.queueID || r != c.recipient || s != c.status {
|
||||
t.Fatalf("got (%q,%q,%q), want (%q,%q,%q)", q, r, s, c.queueID, c.recipient, c.status)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// captureStore records UpdateStatus calls for the follow integration test.
|
||||
type captureStore struct {
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
}
|
||||
|
||||
func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.calls = append(c.calls, queueID+"|"+recipient+"|"+status)
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil }
|
||||
|
||||
func (c *captureStore) snapshot() []string {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return append([]string(nil), c.calls...)
|
||||
}
|
||||
|
||||
// TestFollowTailsAndRotates writes delivery lines to a log file, then rotates
|
||||
// it (rename + fresh create, as logrotate does) and writes more, asserting the
|
||||
// tailer picks up lines from both the original and rotated file.
|
||||
func TestFollowTailsAndRotates(t *testing.T) {
|
||||
old := pollInterval
|
||||
pollInterval = 10 * time.Millisecond
|
||||
t.Cleanup(func() { pollInterval = old })
|
||||
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "mail.log")
|
||||
if err := os.WriteFile(path, []byte("preexisting line, ignored on start\n"), 0o644); err != nil {
|
||||
t.Fatalf("seed log: %v", err)
|
||||
}
|
||||
|
||||
cs := &captureStore{}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- Run(ctx, path, cs, 90) }()
|
||||
|
||||
// Give follow() time to open at EOF (it seeks to end immediately on start,
|
||||
// so the seed line above is ignored), then append a delivery line.
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, dsn=2.0.0, status=sent (ok)")
|
||||
waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") })
|
||||
|
||||
// Rotate: move the current file aside and create a fresh one (logrotate
|
||||
// "create"), then append to the new file.
|
||||
if err := os.Rename(path, path+".1"); err != nil {
|
||||
t.Fatalf("rotate: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(path, nil, 0o644); err != nil {
|
||||
t.Fatalf("recreate: %v", err)
|
||||
}
|
||||
appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, dsn=5.1.1, status=bounced (nope)")
|
||||
waitFor(t, func() bool { return contains(cs.snapshot(), "Q2|b@example.net|bounced") })
|
||||
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Run did not return after cancel")
|
||||
}
|
||||
}
|
||||
|
||||
func appendLine(t *testing.T, path, line string) {
|
||||
t.Helper()
|
||||
f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
t.Fatalf("open for append: %v", err)
|
||||
}
|
||||
defer f.Close()
|
||||
if _, err := f.WriteString(line + "\n"); err != nil {
|
||||
t.Fatalf("append: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func contains(ss []string, want string) bool {
|
||||
for _, s := range ss {
|
||||
if s == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func waitFor(t *testing.T, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("condition not met within timeout")
|
||||
}
|
||||
Reference in New Issue
Block a user