119 lines
3.7 KiB
Go
119 lines
3.7 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// syncRecorder is a ResponseRecorder safe to read while the handler writes.
|
|
type syncRecorder struct {
|
|
*httptest.ResponseRecorder
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func (r *syncRecorder) Write(b []byte) (int, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.ResponseRecorder.Write(b)
|
|
}
|
|
|
|
func (r *syncRecorder) body() string {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.Body.String()
|
|
}
|
|
|
|
// Last-Event-ID 0-0 on meta:events returns the whole current state — one set
|
|
// per record key, only the last carrying an id — then the live tail.
|
|
func TestMetaEventsSnapshotThenTail(t *testing.T) {
|
|
stor := newTestStorage(t)
|
|
s := &Server{storage: stor}
|
|
a := testSigner(t, 1)
|
|
for i, cid := range []string{"bafyR1", "bafyR2"} {
|
|
storeClaims(t, stor, signed(t, a, cid, "title", "T", 1), signed(t, a, cid, "n", string(rune('a'+i)), 1))
|
|
}
|
|
r1, r2 := stor.ResolveRoot("bafyR1"), stor.ResolveRoot("bafyR2")
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
req := httptest.NewRequest(http.MethodGet, "/api/events/meta", nil).WithContext(ctx)
|
|
req.Header.Set("Last-Event-ID", "0-0")
|
|
rec := &syncRecorder{ResponseRecorder: httptest.NewRecorder()}
|
|
done := make(chan struct{})
|
|
go func() { s.handleEventsMeta(rec, req); close(done) }()
|
|
|
|
waitFor := func(substr string) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for !strings.Contains(rec.body(), substr) {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("timed out waiting for %q in:\n%s", substr, rec.body())
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
}
|
|
waitFor(`"key":"file:` + r2 + `/title"`)
|
|
storeClaims(t, stor, signed(t, a, "bafyR1", "title", "changed", 2))
|
|
waitFor(`event: set`)
|
|
time.Sleep(300 * time.Millisecond)
|
|
cancel()
|
|
<-done
|
|
|
|
body := rec.body()
|
|
keys := []string{"file:" + r1 + "/title", "file:" + r1 + "/n", "file:" + r2 + "/title", "file:" + r2 + "/n",
|
|
"file:" + r1 + "/cids/bafyR1", "file:" + r2 + "/cids/bafyR2"}
|
|
for _, k := range keys {
|
|
if !strings.Contains(body, `"key":"`+k+`"`) {
|
|
t.Errorf("snapshot is missing %s", k)
|
|
}
|
|
}
|
|
events := strings.Split(strings.TrimSpace(body), "\n\n")
|
|
if len(events) != 7 {
|
|
t.Fatalf("want 6 snapshot events + 1 live event, got %d:\n%s", len(events), body)
|
|
}
|
|
for i, e := range events[:5] {
|
|
if strings.Contains(e, "id: ") {
|
|
t.Errorf("snapshot event %d carries an id: %q", i, e)
|
|
}
|
|
}
|
|
if !strings.HasPrefix(events[5], "id: ") || !strings.HasPrefix(events[6], "id: ") {
|
|
t.Fatalf("the last snapshot event and the live event must carry ids:\n%s", body)
|
|
}
|
|
if !strings.Contains(events[6], `"key":"file:`+r1+`/title"`) {
|
|
t.Fatalf("live tail did not follow the snapshot: %q", events[6])
|
|
}
|
|
}
|
|
|
|
// A cursor older than the last cache rebuild gets `gap` (reason rebuild): the
|
|
// rebuild published no per-key events, so the consumer must re-snapshot.
|
|
func TestMetaEventsGapAfterRebuild(t *testing.T) {
|
|
stor := newTestStorage(t)
|
|
s := &Server{storage: stor}
|
|
storeClaims(t, stor, signed(t, testSigner(t, 1), "bafyA", "title", "T", 1))
|
|
if err := stor.Rebuild(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
req := httptest.NewRequest(http.MethodGet, "/api/events/meta", nil).WithContext(ctx)
|
|
req.Header.Set("Last-Event-ID", "1000-0")
|
|
rec := &syncRecorder{ResponseRecorder: httptest.NewRecorder()}
|
|
done := make(chan struct{})
|
|
go func() { s.handleEventsMeta(rec, req); close(done) }()
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for !strings.Contains(rec.body(), "event: gap") {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("no gap after rebuild:\n%s", rec.body())
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
cancel()
|
|
<-done
|
|
if !strings.Contains(rec.body(), `"reason":"rebuild"`) {
|
|
t.Fatalf("gap reason: %s", rec.body())
|
|
}
|
|
}
|