1package daemon23import (4 "context"5 "encoding/json"6 "errors"7 "io"8 "net/http"9 "net/http/httptest"10 "strings"11 "sync"12 "testing"13 "time"1415 "github.com/FelineStateMachine/tinyrelay/internal/event"16 "github.com/FelineStateMachine/tinyrelay/internal/relay"17 "github.com/FelineStateMachine/tinyrelay/internal/replication"18)1920type waitingRequestBody struct {21 reader io.Reader22 entered chan struct{}23 release chan struct{}24 once sync.Once25}2627func (b *waitingRequestBody) Read(p []byte) (int, error) {28 b.once.Do(func() { close(b.entered); <-b.release })29 return b.reader.Read(p)30}31func (b *waitingRequestBody) Close() error { return nil }3233func TestBackupDrainsActualHTTPPublishAndResumes(t *testing.T) {34 _, tenant := testTenant(t)35 e := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "publish at snapshot boundary"}36 if err := event.Sign(&e, strings.Repeat("0", 63)+"1"); err != nil {37 t.Fatal(err)38 }39 raw, err := json.Marshal(e)40 if err != nil {41 t.Fatal(err)42 }43 body := &waitingRequestBody{reader: strings.NewReader(string(raw)), entered: make(chan struct{}), release: make(chan struct{})}44 req := httptest.NewRequest(http.MethodPost, "http://relay.test/events", body)45 signRequest(t, req, string(raw))46 res := httptest.NewRecorder()47 published := make(chan struct{})48 go func() { tenant.ServeHTTP(res, req); close(published) }()49 <-body.entered50 type snapshotResult struct {51 archive replication.BackupArchive52 err error53 }54 snapshot := make(chan snapshotResult, 1)55 go func() {56 archive, err := replication.CreateBackupWithProvider(context.Background(), tenant.store, tenant.ReplicationBackupProvider(), time.Now().Unix())57 snapshot <- snapshotResult{archive, err}58 }()59 deadline := time.After(time.Second)60 for {61 tenant.maintenance.mu.Lock()62 exclusive := tenant.maintenance.maintenance63 tenant.maintenance.mu.Unlock()64 if exclusive {65 break66 }67 select {68 case <-deadline:69 close(body.release)70 t.Fatal("snapshot did not request maintenance")71 case <-time.After(time.Millisecond):72 }73 }74 select {75 case <-snapshot:76 close(body.release)77 t.Fatal("snapshot passed in-flight upload")78 default:79 }80 if _, err := tenant.Query(context.Background(), []event.Filter{{}}, relay.Session{}); !errors.Is(err, ErrMaintenance) {81 close(body.release)82 t.Fatalf("new query during snapshot: %v", err)83 }84 close(body.release)85 <-published86 if res.Code != http.StatusOK {87 t.Fatalf("in-flight publish %d: %s", res.Code, res.Body.String())88 }89 result := <-snapshot90 if result.err != nil {91 t.Fatal(result.err)92 }93 found := false94 for _, item := range result.archive.Events {95 if item.ID == e.ID {96 found = true97 }98 }99 if !found {100 t.Fatal("snapshot lost the in-flight committed event")101 }102 if _, err := tenant.Query(context.Background(), []event.Filter{{IDs: []string{e.ID}}}, relay.Session{}); err != nil {103 t.Fatalf("query after snapshot: %v", err)104 }105}106107func TestShutdownWaitsForExclusiveMaintenance(t *testing.T) {108 var gate maintenanceGate109 _, release, err := gate.beginMaintenance(context.Background())110 if err != nil {111 t.Fatal(err)112 }113 gate.markClosing()114 idle := make(chan error, 1)115 go func() { idle <- gate.waitIdle(context.Background()) }()116 select {117 case err := <-idle:118 t.Fatalf("shutdown passed active maintenance: %v", err)119 case <-time.After(20 * time.Millisecond):120 }121 release()122 select {123 case err := <-idle:124 if err != nil {125 t.Fatal(err)126 }127 case <-time.After(time.Second):128 t.Fatal("shutdown did not resume")129 }130}