amadan

carlos / platform Public

host: ship coalesced errors like logs; prune by chunk family

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 2a3756f8160089b3c21846048a4a5b65933ff6ed parent 5e88ea1
5 files changed, +161 −31
  • internal/host/agent.go +6 −0
  • internal/host/errship.go +109 −0
  • internal/host/errship_test.go +37 −0
  • internal/host/logship.go +7 −22
  • internal/host/logship_test.go +2 −9
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 0208f0dc7..04e45887d 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -166,6 +166,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/apperr"
"github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/caps"
"github.com/carlosframework/platform/internal/cgroup"
@@ -232,8 +233,10 @@ func RunAgent(args []string) {
// it away is deliberate and cheap (neither branch makes a call), and it
// asks precisely the question logshipLoop will ask again for itself.
var logSink *applog.Sink
+ var errSink *apperr.Sink
if _, _, err := openStatusStore(context.Background(), os.Getenv); err == nil {
logSink = applog.NewSink(status.Label(readMachineID()), time.Now)
+ errSink = apperr.NewSink(status.Label(readMachineID()), time.Now)
}
cfg, err := activator.ConfigFromEnv()
@@ -383,6 +386,9 @@ func RunAgent(args []string) {
if logSink != nil {
go logshipLoop(logSink, status.Label(readMachineID()))
}
+ if errSink != nil {
+ go errshipLoop(errSink, status.Label(readMachineID()))
+ }
go statusLoop(reg)
// The memory early-warning (memory-pressure spec §2.4): per box, gated
// by CARLOS_HOST_METRICS — one log line and nothing else where unset.
diff --git a/internal/host/errship.go b/internal/host/errship.go
new file mode 100644
index 000000000..005a936b7
--- /dev/null
+++ b/internal/host/errship.go
@@ -0,0 +1,109 @@
+// The error shipping loop — the box-side pump of the Errors tab (spec
+// 2026-09-07-errors-and-analytics-design.md §3.2), shaped exactly like
+// logshipLoop: the edge and the activator coalesce into one apperr.Sink;
+// every tick this drains it and writes one immutable chunk per app under
+// apps/<sqid>/<app>/errors/<box>/<hour>/; on each hour boundary it prunes
+// its own subtree with the log retention (§3.6: errors follow log
+// retention, age and byte cap).
+package host
+
+import (
+ "context"
+ "errors"
+ "log"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apperr"
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const errshipTick = 10 * time.Second
+
+func errshipLoop(sink *apperr.Sink, box string) {
+ defer func() {
+ if r := recover(); r != nil {
+ log.Printf("agent: errship loop panic, error shipping stopped for this process: %v", r)
+ }
+ }()
+ ctx := context.Background()
+ s, target, err := openStatusStore(ctx, os.Getenv)
+ if err != nil {
+ if errors.Is(err, errNoStatusStore) {
+ log.Printf("agent: errship loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
+ } else {
+ log.Printf("agent: errship loop disabled: %v", err)
+ }
+ return
+ }
+ log.Printf("agent: errship loop armed (10s ticks), publishing to %s", target)
+
+ pending := make(map[applog.AppKey][]apperr.Record)
+ pruned := ""
+ ticker := time.NewTicker(errshipTick)
+ defer ticker.Stop()
+ for range ticker.C {
+ errshipGather(sink, pending)
+ errshipFlush(ctx, s, pending)
+ if hour := applog.HourOf(time.Now()); hour != pruned {
+ for _, k := range errshipKnownApps(pending, sink) {
+ chunkPrune(ctx, s, applog.FamilyErrors, box, k, time.Now())
+ }
+ pruned = hour
+ }
+ }
+}
+
+// errshipGather drains the sink into pending under the same per-app cap
+// logship uses — oldest first.
+func errshipGather(sink *apperr.Sink, pending map[applog.AppKey][]apperr.Record) {
+ for k, recs := range sink.Drain() {
+ merged := append(pending[k], recs...)
+ if over := len(merged) - logshipMaxPending; over > 0 {
+ merged = merged[over:]
+ }
+ pending[k] = merged
+ }
+}
+
+// errshipFlush writes one create-only chunk per pending app and forgets
+// what landed; a failed app keeps its records for the next round.
+func errshipFlush(ctx context.Context, s release.Store, pending map[applog.AppKey][]apperr.Record) {
+ for k, recs := range pending {
+ if len(recs) == 0 {
+ delete(pending, k)
+ continue
+ }
+ body, err := apperr.EncodeChunk(recs)
+ if err != nil {
+ log.Printf("agent: errship encode %s/%s: %v (dropping %d records)", k.Acct, k.App, err, len(recs))
+ delete(pending, k)
+ continue
+ }
+ key := apperr.ChunkKey(k.Acct, k.App, recs[0].Box, apperr.HourOf(recs[0].TS), recs[0].Seq, len(body))
+ if _, err := s.Put(ctx, key, body, ""); err != nil {
+ log.Printf("agent: errship flush %s/%s: %v (%d records held)", k.Acct, k.App, err, len(recs))
+ continue
+ }
+ delete(pending, k)
+ }
+}
+
+func errshipKnownApps(pending map[applog.AppKey][]apperr.Record, sink *apperr.Sink) []applog.AppKey {
+ seen := make(map[applog.AppKey]bool)
+ var out []applog.AppKey
+ for k := range pending {
+ if !seen[k] {
+ seen[k] = true
+ out = append(out, k)
+ }
+ }
+ for _, k := range sink.Known() {
+ if !seen[k] {
+ seen[k] = true
+ out = append(out, k)
+ }
+ }
+ return out
+}
diff --git a/internal/host/errship_test.go b/internal/host/errship_test.go
new file mode 100644
index 000000000..83a7e033e
--- /dev/null
+++ b/internal/host/errship_test.go
@@ -0,0 +1,37 @@
+package host
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apperr"
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestErrshipFlushWritesOneChunkPerApp(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ now := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC)
+ sink := apperr.NewSink("box1", func() time.Time { return now })
+ sink.Add("bab", "web", apperr.Record{Source: apperr.SourceEdge, Kind: apperr.KindHTTP, Name: "500 GET /x", Host: "h1"})
+ sink.Add("bab", "web", apperr.Record{Source: apperr.SourceEdge, Kind: apperr.KindHTTP, Name: "500 GET /x", Host: "h1"})
+ pending := map[applog.AppKey][]apperr.Record{}
+ errshipGather(sink, pending)
+ errshipFlush(context.Background(), s, pending)
+ if len(pending) != 0 {
+ t.Fatalf("pending after flush = %d", len(pending))
+ }
+ recs, _, err := apperr.ReadWindow(context.Background(), s, "bab", "web", apperr.Filter{Since: now.Add(-time.Minute)}, 100)
+ if err != nil || len(recs) != 1 || recs[0].Count != 2 {
+ t.Fatalf("read back = %+v, %v", recs, err)
+ }
+}
+
+func TestErrshipFlushHoldsOnFailure(t *testing.T) {
+ pending := map[applog.AppKey][]apperr.Record{{Acct: "bab", App: "web"}: {{TS: time.Now(), Source: apperr.SourceEdge, Name: "x", FP: "f", Box: "box1", Seq: 1}}}
+ errshipFlush(context.Background(), failingStore{}, pending)
+ if len(pending) != 1 {
+ t.Fatal("a failed flush must keep its records for the next round")
+ }
+}
diff --git a/internal/host/logship.go b/internal/host/logship.go
index 1f097be4d..a0edab3b3 100644
--- a/internal/host/logship.go
+++ b/internal/host/logship.go
@@ -89,7 +89,7 @@ func logshipLoop(sink *applog.Sink, box string) {
// other boxes' passes or the app's own deletion sweep. Sub-cap
// apps pay one ListDirs each per hour.
for _, k := range logshipKnownApps(pending, sink) {
- logshipPrune(ctx, s, box, k, time.Now())
+ chunkPrune(ctx, s, applog.FamilyLogs, box, k, time.Now())
}
pruned = hour
}
@@ -153,13 +153,13 @@ func logshipKnownApps(pending map[applog.AppKey][]applog.Record, sink *applog.Si
return out
}
-// logshipPrune enforces retention for ONE app on THIS box's subtree only —
-// each box owns what it wrote (spec §5). Hour granularity: an hour goes
-// when it is older than the age floor, or (oldest first) while the
+// chunkPrune enforces retention for ONE app's ONE family on THIS box's
+// subtree only — each box owns what it wrote (spec §5). Hour granularity: an
+// hour goes when it is older than the age floor, or (oldest first) while the
// subtree's key-embedded sizes total over the byte cap. Errors are logged
// and left for the next hourly pass — retention is idempotent.
-func logshipPrune(ctx context.Context, s release.Store, box string, k applog.AppKey, now time.Time) {
- prefix := applog.BoxPrefix(k.Acct, k.App, box)
+func chunkPrune(ctx context.Context, s release.Store, fam applog.Family, box string, k applog.AppKey, now time.Time) {
+ prefix := applog.FamilyBoxPrefix(fam, k.Acct, k.App, box)
hourDirs, err := s.ListDirs(ctx, prefix)
if err != nil {
log.Printf("agent: logship prune %s: %v", prefix, err)
@@ -195,7 +195,7 @@ func logshipPrune(ctx context.Context, s release.Store, box string, k applog.App
// hours[0] is oldest.
cutoff := now.Add(-logshipMaxAge)
for _, h := range hours {
- hourName := hourNameOf(h.dir)
+ hourName := applog.HourNameOfDir(h.dir)
drop := applog.HourBefore(hourName, cutoff) || total > logshipMaxBytes
if !drop {
break // hours are oldest-first; nothing further is droppable
@@ -209,18 +209,3 @@ func logshipPrune(ctx context.Context, s release.Store, box string, k applog.App
total -= h.bytes
}
}
-
-// hourNameOf extracts the hour directory name from its full prefix
-// (".../logs/<box>/<hour>/" → "<hour>").
-func hourNameOf(dir string) string {
- trimmed := dir
- if n := len(trimmed); n > 0 && trimmed[n-1] == '/' {
- trimmed = trimmed[:n-1]
- }
- for i := len(trimmed) - 1; i >= 0; i-- {
- if trimmed[i] == '/' {
- return trimmed[i+1:]
- }
- }
- return trimmed
-}
diff --git a/internal/host/logship_test.go b/internal/host/logship_test.go
index 4712d30e8..331a7c62e 100644
--- a/internal/host/logship_test.go
+++ b/internal/host/logship_test.go
@@ -98,7 +98,7 @@ func TestLogshipPrune(t *testing.T) {
liveKey := put("box1", "2026-08-10-11", 3, 10)
otherBox := put("box2", "2026-08-01-10", 4, 10) // NOT ours to prune
- logshipPrune(ctx, s, "box1", k, now)
+ chunkPrune(ctx, s, applog.FamilyLogs, "box1", k, now)
for key, want := range map[string]bool{oldKey: false, bigKey: false, liveKey: true, otherBox: true} {
_, _, err := s.Get(ctx, key)
@@ -115,13 +115,6 @@ func (failingStore) Put(context.Context, string, []byte, string) (string, error)
return "", context.DeadlineExceeded
}
-// TestHourNameOf pins the prefix→name extraction.
-func TestHourNameOf(t *testing.T) {
- if got := hourNameOf("apps/bde/ledgerpay/logs/box1/2026-08-10-11/"); got != "2026-08-10-11" {
- t.Fatalf("hourNameOf = %q", got)
- }
-}
-
// TestLogshipPruneKeepsRecentUnderCap: nothing under both limits is touched.
func TestLogshipPruneKeepsRecentUnderCap(t *testing.T) {
s := release.NewDirStore(t.TempDir())
@@ -132,7 +125,7 @@ func TestLogshipPruneKeepsRecentUnderCap(t *testing.T) {
if _, err := s.Put(ctx, key, make([]byte, 10), ""); err != nil {
t.Fatal(err)
}
- logshipPrune(ctx, s, "box1", k, now)
+ chunkPrune(ctx, s, applog.FamilyLogs, "box1", k, now)
if _, _, err := s.Get(ctx, key); err != nil {
t.Fatalf("live chunk pruned: %v", err)
}