carlos / platform Public

Download this file

1	package host
2
3	import (
4	"context"
5	"strings"
6	"testing"
7	"time"
8
9	"github.com/carlosframework/platform/internal/applog"
10	"github.com/carlosframework/platform/internal/release"
11	)
12
13	func logshipFixture(t *testing.T) (release.Store, *applog.Sink) {
14	t.Helper()
15	s := release.NewDirStore(t.TempDir())
16	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
17	return s, applog.NewSink("box1", func() time.Time { return now })
18	}
19
20	// TestLogshipFlushWritesOneChunkPerApp: a drain with two apps lands as two
21	// immutable chunks, each under its own app prefix, decodable and complete.
22	func TestLogshipFlushWritesOneChunkPerApp(t *testing.T) {
23	s, sink := logshipFixture(t)
24	sink.App("bde", "ledgerpay", "ledgerpay.bde.oncarlos.com", "hello")
25	sink.Event("bab", "kass", "kass.example.com", applog.LevelInfo, "instance woke")
26
27	pending := make(map[applog.AppKey][]applog.Record)
28	logshipGather(sink, pending)
29	logshipFlush(context.Background(), s, pending)
30	if len(pending) != 0 {
31	t.Fatalf("flushed apps still pending: %v", pending)
32	}
33
34	for _, app := range []applog.AppKey{{Acct: "bde", App: "ledgerpay"}, {Acct: "bab", App: "kass"}} {
35	keys, err := s.List(context.Background(), mustKey(t)(applog.Prefix(app.Acct, app.App)))
36	if err != nil || len(keys) != 1 {
37	t.Fatalf("chunks for %v: %v (%v)", app, keys, err)
38	}
39	body, _, err := s.Get(context.Background(), keys[0])
40	if err != nil {
41	t.Fatal(err)
42	}
43	recs, err := applog.DecodeChunk(body)
44	if err != nil || len(recs) != 1 {
45	t.Fatalf("chunk decode for %v: %d recs, %v", app, len(recs), err)
46	}
47	// The key carries the real byte size — retention counts with it.
48	if n, ok := applog.ChunkKeySize(keys[0]); !ok || n != int64(len(body)) {
49	t.Fatalf("key size %d/%v, body %d: %s", n, ok, len(body), keys[0])
50	}
51	}
52	}
53
54	// TestLogshipFlushKeepsRecordsWhenTheBucketRefuses: a failed Put must hold
55	// the app's records for the next round, not lose them — and the pending cap
56	// bounds that holding.
57	func TestLogshipFlushKeepsRecordsWhenTheBucketRefuses(t *testing.T) {
58	_, sink := logshipFixture(t)
59	sink.App("bde", "ledgerpay", "h", "precious")
60	pending := make(map[applog.AppKey][]applog.Record)
61	logshipGather(sink, pending)
62	logshipFlush(context.Background(), failingStore{}, pending)
63	k := applog.AppKey{Acct: "bde", App: "ledgerpay"}
64	if len(pending[k]) != 1 || pending[k][0].Msg != "precious" {
65	t.Fatalf("records lost on failed flush: %v", pending)
66	}
67	// Cap: a long outage trims oldest, keeps the freshest tail.
68	for i := 0; i < logshipMaxPending+10; i++ {
69	sink.App("bde", "ledgerpay", "h", "line")
70	}
71	logshipGather(sink, pending)
72	if len(pending[k]) != logshipMaxPending {
73	t.Fatalf("pending cap not enforced: %d", len(pending[k]))
74	}
75	if pending[k][0].Msg == "precious" {
76	t.Fatal("cap kept oldest instead of freshest")
77	}
78	}
79
80	// TestLogshipPrune: hours past the age floor go; then oldest-first until
81	// the key-embedded sizes fit the byte cap; the box only touches its own
82	// subtree.
83	func TestLogshipPrune(t *testing.T) {
84	t.Parallel()
85	s := release.NewDirStore(t.TempDir())
86	ctx := context.Background()
87	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
88	k := applog.AppKey{Acct: "bde", App: "ledgerpay"}
89
90	put := func(box, hour string, seq uint64, size int) string {
91	key := mustKey(t)(applog.ChunkKey(k.Acct, k.App, box, hour, seq, size))
92	if _, err := s.Put(ctx, key, make([]byte, size), ""); err != nil {
93	t.Fatal(err)
94	}
95	return key
96	}
97	oldKey := put("box1", "2026-08-01-10", 1, 10) // over the 7d floor
98	bigKey := put("box1", "2026-08-10-10", 2, 60<<20) // pushes total over the byte cap
99	liveKey := put("box1", "2026-08-10-11", 3, 10)
100	otherBox := put("box2", "2026-08-01-10", 4, 10) // NOT ours to prune
101
102	chunkPrune(ctx, s, applog.FamilyLogs, "box1", k, now)
103
104	for key, want := range map[string]bool{oldKey: false, bigKey: false, liveKey: true, otherBox: true} {
105	_, _, err := s.Get(ctx, key)
106	if got := err == nil; got != want {
107	t.Errorf("%s survived=%v, want %v", key, got, want)
108	}
109	}
110	}
111
112	// failingStore refuses writes — the bucket-outage fixture.
113	type failingStore struct{ release.Store }
114
115	func (failingStore) Put(context.Context, string, []byte, string) (string, error) {
116	return "", context.DeadlineExceeded
117	}
118
119	// TestLogshipPruneKeepsRecentUnderCap: nothing under both limits is touched.
120	func TestLogshipPruneKeepsRecentUnderCap(t *testing.T) {
121	s := release.NewDirStore(t.TempDir())
122	ctx := context.Background()
123	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
124	k := applog.AppKey{Acct: "bde", App: "ledgerpay"}
125	key := mustKey(t)(applog.ChunkKey(k.Acct, k.App, "box1", "2026-08-10-11", 1, 10))
126	if _, err := s.Put(ctx, key, make([]byte, 10), ""); err != nil {
127	t.Fatal(err)
128	}
129	chunkPrune(ctx, s, applog.FamilyLogs, "box1", k, now)
130	if _, _, err := s.Get(ctx, key); err != nil {
131	t.Fatalf("live chunk pruned: %v", err)
132	}
133	if !strings.HasPrefix(key, "apps/bde/ledgerpay/logs/") {
134	t.Fatalf("key layout drifted: %s", key)
135	}
136	}
137
138	// must unwraps a key builder's (key, error) for fixtures whose names are
139	// known good: mustKey(t)(ChunkKey(...)).
140	func mustKey(t testing.TB) func(string, error) string {
141	return func(k string, err error) string {
142	t.Helper()
143	if err != nil {
144	t.Fatal(err)
145	}
146	return k
147	}
148	}
149