diff --git a/decision/receipt_export.go b/decision/receipt_export.go index 246f29b..ea3dedc 100644 --- a/decision/receipt_export.go +++ b/decision/receipt_export.go @@ -121,6 +121,9 @@ func (exporter *ReceiptExporter) ExportOnce(ctx context.Context) error { } exporter.mu.Lock() defer exporter.mu.Unlock() + if err := exporter.journal.Refresh(); err != nil { + return fmt.Errorf("decision: refresh receipt journal: %w", err) + } attempted := 0 var firstErr error for _, receipt := range exporter.journal.Receipts() { diff --git a/decision/receipt_export_test.go b/decision/receipt_export_test.go index ca122d8..7514ad3 100644 --- a/decision/receipt_export_test.go +++ b/decision/receipt_export_test.go @@ -7,11 +7,72 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "os" "path/filepath" + "strings" "sync/atomic" "testing" ) +func TestReceiptExporterDiscoversEvidenceAppendedByAnotherProcess(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "receipts.jsonl") + exporterJournal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + writerJournal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + receipt := journalReceipt(t, 1785500000, Enforced) + if err := writerJournal.AppendReceipt(context.Background(), receipt); err != nil { + t.Fatal(err) + } + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + calls.Add(1) + if request.Header.Get("Idempotency-Key") != receipt.ID { + t.Errorf("idempotency key=%q", request.Header.Get("Idempotency-Key")) + } + _ = json.NewEncoder(writer).Encode(map[string]string{"accepted_receipt_id": receipt.ID}) + })) + defer server.Close() + exporter, err := NewReceiptExporter(ReceiptExporterConfig{ + Journal: exporterJournal, Endpoint: server.URL, AckPath: filepath.Join(t.TempDir(), "acks"), + }) + if err != nil { + t.Fatal(err) + } + if err := exporter.ExportOnce(context.Background()); err != nil { + t.Fatal(err) + } + if calls.Load() != 1 || exporter.Pending() != 0 { + t.Fatalf("external receipt calls=%d pending=%d", calls.Load(), exporter.Pending()) + } +} + +func TestReceiptExporterSurfacesIncompleteExternalJournalRecord(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "receipts.jsonl") + journal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + exporter, err := NewReceiptExporter(ReceiptExporterConfig{ + Journal: journal, Endpoint: "http://127.0.0.1:1/receipts", AckPath: filepath.Join(t.TempDir(), "acks"), + }) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("{"), 0o600); err != nil { + t.Fatal(err) + } + if err := exporter.ExportOnce(context.Background()); err == nil || !strings.Contains(err.Error(), "refresh receipt journal") { + t.Fatalf("incomplete external journal error=%v", err) + } +} + func TestReceiptExporterRetriesAndAcknowledgesSignedEvidence(t *testing.T) { t.Parallel() journal, err := OpenReceiptJournal(filepath.Join(t.TempDir(), "receipts.jsonl")) diff --git a/decision/receipt_journal.go b/decision/receipt_journal.go index 69fa2f1..9a3c475 100644 --- a/decision/receipt_journal.go +++ b/decision/receipt_journal.go @@ -27,6 +27,7 @@ type ReceiptJournal struct { mu sync.Mutex seen map[string]string receipts []Receipt + offset int64 } func OpenReceiptJournal(path string) (*ReceiptJournal, error) { @@ -46,7 +47,7 @@ func OpenReceiptJournal(path string) (*ReceiptJournal, error) { return nil, fmt.Errorf("decision: create receipt journal directory: %w", err) } journal := &ReceiptJournal{path: absolute, seen: make(map[string]string)} - if err := journal.load(); err != nil { + if err := journal.refreshLocked(); err != nil { return nil, err } return journal, nil @@ -107,11 +108,31 @@ func (journal *ReceiptJournal) Receipts() []Receipt { return append([]Receipt(nil), journal.receipts...) } -func (journal *ReceiptJournal) load() error { - file, err := os.Open(journal.path) +// Refresh discovers complete records appended by another process after this +// journal was opened. Managed hooks and the long-running daemon deliberately +// use separate processes, so exporters must refresh before inspecting their +// in-memory snapshot. Existing records are not rescanned on every poll. +func (journal *ReceiptJournal) Refresh() error { + if journal == nil || journal.path == "" { + return fmt.Errorf("decision: receipt journal is not initialized") + } + journal.mu.Lock() + defer journal.mu.Unlock() + return journal.refreshLocked() +} + +func (journal *ReceiptJournal) refreshLocked() error { + pathInfo, err := os.Lstat(journal.path) if os.IsNotExist(err) { return nil } + if err != nil { + return fmt.Errorf("decision: inspect receipt journal: %w", err) + } + if pathInfo.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("decision: receipt journal must not be a symlink") + } + file, err := os.Open(journal.path) if err != nil { return fmt.Errorf("decision: open receipt journal: %w", err) } @@ -123,9 +144,28 @@ func (journal *ReceiptJournal) load() error { if info.Mode().Perm()&0o077 != 0 { return fmt.Errorf("decision: receipt journal permissions must be owner-only") } - scanner := bufio.NewScanner(file) + if info.Size() < journal.offset { + return fmt.Errorf("decision: receipt journal was truncated") + } + if info.Size() == journal.offset { + return nil + } + var last [1]byte + if _, err := file.ReadAt(last[:], info.Size()-1); err != nil { + return fmt.Errorf("decision: inspect receipt journal boundary: %w", err) + } + if last[0] != '\n' { + return fmt.Errorf("decision: receipt journal has an incomplete trailing record") + } + scanner := bufio.NewScanner(io.NewSectionReader(file, journal.offset, info.Size()-journal.offset)) scanner.Buffer(make([]byte, 64<<10), MaxReceiptJournalLineBytes+1) - line := 0 + line := len(journal.receipts) + type pendingReceipt struct { + receipt Receipt + hash string + } + pending := make([]pendingReceipt, 0) + pendingSeen := make(map[string]string) for scanner.Scan() { line++ decoder := json.NewDecoder(bytes.NewReader(scanner.Bytes())) @@ -148,12 +188,26 @@ func (journal *ReceiptJournal) load() error { if existing, exists := journal.seen[receipt.ID]; exists && existing != hash { return fmt.Errorf("decision: conflicting receipt journal id %q", receipt.ID) } - journal.seen[receipt.ID] = hash - journal.receipts = append(journal.receipts, receipt) + if existing, exists := pendingSeen[receipt.ID]; exists && existing != hash { + return fmt.Errorf("decision: conflicting receipt journal id %q", receipt.ID) + } + if existing, exists := journal.seen[receipt.ID]; exists && existing == hash { + continue + } + if existing, exists := pendingSeen[receipt.ID]; exists && existing == hash { + continue + } + pendingSeen[receipt.ID] = hash + pending = append(pending, pendingReceipt{receipt: receipt, hash: hash}) } if err := scanner.Err(); err != nil { return fmt.Errorf("decision: scan receipt journal: %w", err) } + for _, record := range pending { + journal.seen[record.receipt.ID] = record.hash + journal.receipts = append(journal.receipts, record.receipt) + } + journal.offset = info.Size() return nil } diff --git a/decision/receipt_journal_test.go b/decision/receipt_journal_test.go index 157fe06..f9d09a7 100644 --- a/decision/receipt_journal_test.go +++ b/decision/receipt_journal_test.go @@ -6,10 +6,14 @@ import ( "context" "crypto/ed25519" "crypto/rand" + "encoding/json" "os" "path/filepath" + "strings" "testing" "time" + + "github.com/pilot-protocol/common/fsutil" ) func journalReceipt(t *testing.T, observedAt int64, result EnforcementResult) Receipt { @@ -107,3 +111,92 @@ func TestReceiptJournalRejectsUnsignedCorruptAndUnsafeFiles(t *testing.T) { t.Fatal("symlink journal was accepted") } } + +func TestReceiptJournalRefreshRejectsUnsafeExternalChanges(t *testing.T) { + t.Parallel() + if err := (*ReceiptJournal)(nil).Refresh(); err == nil { + t.Fatal("nil journal refresh succeeded") + } + t.Run("incomplete record", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "receipts.jsonl") + journal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("{"), 0o600); err != nil { + t.Fatal(err) + } + if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "incomplete trailing record") { + t.Fatalf("incomplete refresh error=%v", err) + } + }) + t.Run("truncated journal", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "receipts.jsonl") + journal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + if err := journal.AppendReceipt(context.Background(), journalReceipt(t, 1785500000, Enforced)); err != nil { + t.Fatal(err) + } + if err := journal.Refresh(); err != nil { + t.Fatal(err) + } + if err := os.Truncate(path, 0); err != nil { + t.Fatal(err) + } + if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "truncated") { + t.Fatalf("truncated refresh error=%v", err) + } + }) + t.Run("symlink replacement", func(t *testing.T) { + directory := t.TempDir() + path := filepath.Join(directory, "receipts.jsonl") + journal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + target := filepath.Join(directory, "target.jsonl") + if err := os.WriteFile(target, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(target, path); err != nil { + t.Fatal(err) + } + if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "symlink") { + t.Fatalf("symlink refresh error=%v", err) + } + }) +} + +func TestReceiptJournalRefreshRejectsConflictingExternalRecord(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "receipts.jsonl") + journal, err := OpenReceiptJournal(path) + if err != nil { + t.Fatal(err) + } + receipt := journalReceipt(t, 1785500000, Enforced) + if err := journal.AppendReceipt(context.Background(), receipt); err != nil { + t.Fatal(err) + } + if err := journal.Refresh(); err != nil { + t.Fatal(err) + } + conflict := receipt + conflict.ObservedAt++ + _, privateKey, _ := ed25519.GenerateKey(rand.Reader) + if err := conflict.Sign(privateKey); err != nil { + t.Fatal(err) + } + body, err := json.Marshal(conflict) + if err != nil { + t.Fatal(err) + } + if err := fsutil.AppendSync(path, append(body, '\n')); err != nil { + t.Fatal(err) + } + if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "conflicting receipt journal id") { + t.Fatalf("conflicting refresh error=%v", err) + } +}