tinyrelaycommit

Commit f9b06b9bec64

browse files at this commit | history

1diff --git a/internal/daemon/git_history.go b/internal/daemon/git_history.go
2new file mode 100644
3index 0000000..ac92147
4--- /dev/null
5+++ b/internal/daemon/git_history.go
6@@ -0,0 +1,34 @@
7+package daemon
8+
9+import (
10+ "context"
11+ "errors"
12+ "time"
13+)
14+
15+// Network synchronization has its own lifecycle so an unavailable peer does
16+// not delay local retention, view publication or upload cleanup.
17+func (t *Tenant) runGitHistory(ctx context.Context) {
18+ timer := time.NewTimer(time.Second)
19+ defer timer.Stop()
20+ for {
21+ select {
22+ case <-ctx.Done():
23+ return
24+ case <-timer.C:
25+ }
26+ if t.gitLiveEnabled() {
27+ opCtx, done, err := t.beginOperation(ctx)
28+ if err == nil {
29+ passCtx, cancel := context.WithTimeout(opCtx, 5*time.Minute)
30+ err = t.git.GRASPService().Tick(passCtx)
31+ cancel()
32+ done()
33+ }
34+ if err != nil && !errors.Is(err, context.Canceled) {
35+ t.app.telemetry.Logger().Error("Git synchronization failed", "tenant", t.meta.Name, "error", err)
36+ }
37+ }
38+ timer.Reset(time.Minute)
39+ }
40+}
41diff --git a/internal/daemon/git_live.go b/internal/daemon/git_live.go
42new file mode 100644
43index 0000000..dcbc192
44--- /dev/null
45+++ b/internal/daemon/git_live.go
46@@ -0,0 +1,305 @@
47+package daemon
48+
49+import (
50+ "context"
51+ "errors"
52+ "fmt"
53+ "sort"
54+ "strings"
55+ "time"
56+
57+ "github.com/FelineStateMachine/tinyrelay/internal/event"
58+ "github.com/FelineStateMachine/tinyrelay/internal/gitrelay"
59+ "github.com/FelineStateMachine/tinyrelay/internal/replication"
60+ "github.com/FelineStateMachine/tinyrelay/internal/storage"
61+)
62+
63+const maxGitLivePeers = 32
64+
65+// maxGitLiveRoots bounds the conversation roots one live subscription follows.
66+const maxGitLiveRoots = 100
67+
68+type gitLiveWorker struct {
69+ cancel context.CancelFunc
70+ done chan struct{}
71+ started time.Time
72+}
73+type gitLiveTarget struct {
74+ repo gitrelay.Repository
75+ target string
76+}
77+
78+// runGitLive reconciles the current repository snapshot with live workers. A
79+// worker is removed as soon as its repository, relay, feature or private-peer
80+// authorization disappears. Rotation gives relays beyond the socket budget a
81+// turn on every reconciliation pass.
82+func (t *Tenant) runGitLive(ctx context.Context) {
83+ workers := make(map[string]gitLiveWorker)
84+ var cursor int
85+ defer stopGitLiveWorkers(workers)
86+ for {
87+ if !t.gitLiveEnabled() {
88+ stopGitLiveWorkers(workers)
89+ workers = make(map[string]gitLiveWorker)
90+ } else {
91+ cursor = t.reconcileGitLive(ctx, workers, cursor)
92+ }
93+ timer := time.NewTimer(5 * time.Second)
94+ select {
95+ case <-ctx.Done():
96+ timer.Stop()
97+ return
98+ case <-timer.C:
99+ }
100+ }
101+}
102+
103+func (t *Tenant) gitLiveEnabled() bool {
104+ p := t.Policy()
105+ return p.Features.Grasp && p.Features.Grasp02 && t.git != nil
106+}
107+
108+func (t *Tenant) reconcileGitLive(ctx context.Context, workers map[string]gitLiveWorker, cursor int) int {
109+ desired := t.gitLiveTargets(ctx)
110+ keys := make([]string, 0, len(desired))
111+ for key := range desired {
112+ keys = append(keys, key)
113+ }
114+ sort.Strings(keys)
115+ if len(keys) == 0 {
116+ stopGitLiveWorkers(workers)
117+ return 0
118+ }
119+ if cursor >= len(keys) {
120+ cursor = 0
121+ }
122+ allowed := gitLiveCohort(keys, cursor)
123+ now := time.Now()
124+ for key, worker := range workers {
125+ _, wanted := desired[key]
126+ _, inCohort := allowed[key]
127+ if !wanted || (!inCohort && now.Sub(worker.started) >= gitLiveDwell) {
128+ worker.cancel()
129+ if worker.done != nil {
130+ <-worker.done
131+ }
132+ delete(workers, key)
133+ }
134+ }
135+ allowedKeys := make([]string, 0, len(allowed))
136+ for key := range allowed {
137+ allowedKeys = append(allowedKeys, key)
138+ }
139+ sort.Strings(allowedKeys)
140+ for _, key := range allowedKeys {
141+ if len(workers) >= maxGitLivePeers {
142+ break
143+ }
144+ if _, ok := workers[key]; ok {
145+ continue
146+ }
147+ workerCtx, cancel := context.WithCancel(ctx)
148+ done := make(chan struct{})
149+ workers[key] = gitLiveWorker{cancel: cancel, done: done, started: now}
150+ go func(repo gitrelay.Repository, target string) {
151+ defer close(done)
152+ t.gitLivePeer(workerCtx, repo, target)
153+ }(desired[key].repo, desired[key].target)
154+ }
155+ if len(keys) > maxGitLivePeers {
156+ return (cursor + maxGitLivePeers) % len(keys)
157+ }
158+ return cursor
159+}
160+
161+func gitLiveCohort(keys []string, cursor int) map[string]struct{} {
162+ allowed := make(map[string]struct{}, maxGitLivePeers)
163+ if len(keys) == 0 {
164+ return allowed
165+ }
166+ cursor %= len(keys)
167+ if cursor < 0 {
168+ cursor += len(keys)
169+ }
170+ for i := 0; i < len(keys) && i < maxGitLivePeers; i++ {
171+ allowed[keys[(cursor+i)%len(keys)]] = struct{}{}
172+ }
173+ return allowed
174+}
175+
176+const gitLiveDwell = 2 * time.Minute
177+
178+func (t *Tenant) gitLiveTargets(ctx context.Context) map[string]gitLiveTarget {
179+ result := make(map[string]gitLiveTarget)
180+ for _, repo := range t.git.Repositories() {
181+ targets := append([]string(nil), repo.Relays...)
182+ if t.Policy().Features.Grasp03 {
183+ if authors, err := t.gitLiveConversationAuthors(ctx, repo); err == nil {
184+ offset := t.gitOutboxOffset(ctx, repo)
185+ authors = rotateGitAuthors(authors, offset)
186+ if lists, err := t.gitLocalOutboxEvents(ctx, authors); err == nil {
187+ targets = append(targets, gitOutboxRelays(lists, repo.Private, t.Policy().PrivatePeers)...)
188+ }
189+ }
190+ }
191+ for _, target := range targets {
192+ target = normalizeRelayGit(target)
193+ if target == "" || sameRelayGit(target, t.RelayURL(), t.publicURL) || !t.gitLiveAuthorized(repo, target) {
194+ continue
195+ }
196+ key := strings.Join([]string{repo.Owner, repo.Identifier, target, repo.EventID, fmt.Sprint(repo.Private)}, "\x00")
197+ result[key] = gitLiveTarget{repo: repo, target: target}
198+ }
199+ }
200+ return result
201+}
202+
203+func (t *Tenant) gitLiveConversationAuthors(ctx context.Context, repo gitrelay.Repository) ([]string, error) {
204+ items := make([]event.Event, 0, 256)
205+ for _, filter := range gitEventFilters(repo)[1:] {
206+ rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 256})
207+ if err != nil {
208+ return nil, err
209+ }
210+ items = append(items, rows.Events...)
211+ }
212+ return gitConversationAuthors(items, repo.Owner, repo.Maintainers), nil
213+}
214+
215+func rotateGitAuthors(authors []string, offset int) []string {
216+ if len(authors) == 0 {
217+ return nil
218+ }
219+ offset %= len(authors)
220+ if offset < 0 {
221+ offset += len(authors)
222+ }
223+ rotated := append([]string(nil), authors[offset:]...)
224+ return append(rotated, authors[:offset]...)
225+}
226+
227+func (t *Tenant) gitLiveAuthorized(repo gitrelay.Repository, target string) bool {
228+ if !repo.Private && !privatePolicy(t.Policy()) {
229+ return true
230+ }
231+ return privatePeerMatch(target, t.Policy().PrivatePeers)
232+}
233+
234+func stopGitLiveWorkers(workers map[string]gitLiveWorker) {
235+ for key, worker := range workers {
236+ worker.cancel()
237+ if worker.done != nil {
238+ <-worker.done
239+ }
240+ delete(workers, key)
241+ }
242+}
243+
244+// gitLivePeer performs bounded history overlap before opening subscriptions.
245+// One subscription follows all repository and conversation filters.
246+func (t *Tenant) gitLivePeer(ctx context.Context, repo gitrelay.Repository, target string) {
247+ backoff := time.Second
248+ for ctx.Err() == nil {
249+ historyCtx, historyCancel := context.WithTimeout(ctx, 20*time.Second)
250+ if err := t.gitSyncPeer(historyCtx, repo, target); err != nil && ctx.Err() == nil {
251+ t.app.telemetry.Logger().Debug("Git live history overlap incomplete", "target", target, "error", err)
252+ }
253+ historyCancel()
254+ conversation, err := t.gitConversationEvents(ctx, repo)
255+ if err != nil {
256+ conversation = nil
257+ }
258+ filters := gitLiveFiltersForRoots(repo, conversation)
259+ readyCtx, readyCancel := context.WithTimeout(ctx, 5*time.Second)
260+ transport, err := t.gitTargetTransport(readyCtx, repo, target)
261+ readyCancel()
262+ if err != nil {
263+ backoff = nextGitLiveBackoff(backoff)
264+ if !waitGitLive(ctx, backoff) {
265+ return
266+ }
267+ continue
268+ }
269+ started := time.Now()
270+ subCtx, subCancel := context.WithTimeout(ctx, 2*time.Minute)
271+ err = transport.SubscribeFilters(subCtx, target, filters, func(item event.Event) error {
272+ t.ingestGitLive(ctx, item)
273+ return nil
274+ })
275+ subCancel()
276+ if ctx.Err() != nil {
277+ return
278+ }
279+ if err != nil {
280+ if time.Since(started) >= 30*time.Second {
281+ backoff = time.Second
282+ }
283+ backoff = nextGitLiveBackoff(backoff)
284+ if !waitGitLive(ctx, backoff) {
285+ return
286+ }
287+ }
288+ }
289+}
290+
291+func (t *Tenant) ingestGitLive(ctx context.Context, item event.Event) {
292+ err := t.ingest(ctx, item, replication.OriginImport)
293+ if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) {
294+ return
295+ }
296+}
297+
298+func gitLiveFiltersForRoots(repo gitrelay.Repository, conversation []event.Event) []event.Filter {
299+ filters := gitEventFilters(repo)
300+ roots := make([]string, 0, len(conversation))
301+ for _, item := range conversation {
302+ if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE {
303+ roots = append(roots, item.ID)
304+ }
305+ }
306+ if len(roots) == 0 {
307+ return filters
308+ }
309+ // One subscription carries every filter, so the request stays bounded to
310+ // the newest conversations. Older threads are covered by history passes.
311+ if len(roots) > maxGitLiveRoots {
312+ sorted := append([]event.Event(nil), conversation...)
313+ sortEventsNewestFirst(sorted)
314+ roots = roots[:0]
315+ for _, item := range sorted {
316+ if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE {
317+ roots = append(roots, item.ID)
318+ }
319+ if len(roots) == maxGitLiveRoots {
320+ break
321+ }
322+ }
323+ }
324+ kinds := []int{event.KIND_GIT_PR_UPDATE, 1111, 1630, 1631, 1632, 1633}
325+ return append(filters,
326+ event.Filter{Kinds: kinds, Tags: map[string][]string{"e": roots}},
327+ event.Filter{Kinds: kinds, Tags: map[string][]string{"E": roots}},
328+ )
329+}
330+
331+func nextGitLiveBackoff(current time.Duration) time.Duration {
332+ if current >= 5*time.Minute {
333+ return 5 * time.Minute
334+ }
335+ return current * 2
336+}
337+
338+func waitGitLive(ctx context.Context, delay time.Duration) bool {
339+ timer := time.NewTimer(delay)
340+ defer timer.Stop()
341+ select {
342+ case <-ctx.Done():
343+ return false
344+ case <-timer.C:
345+ return true
346+ }
347+}
348+
349+func liveFilter(repo gitrelay.Repository) event.Filter {
350+ return gitEventFilters(repo)[1]
351+}
352diff --git a/internal/daemon/git_live_test.go b/internal/daemon/git_live_test.go
353new file mode 100644
354index 0000000..e76804b
355--- /dev/null
356+++ b/internal/daemon/git_live_test.go
357@@ -0,0 +1,124 @@
358+package daemon
359+
360+import (
361+ "context"
362+ "strings"
363+ "testing"
364+ "time"
365+
366+ "github.com/FelineStateMachine/tinyrelay/internal/event"
367+ "github.com/FelineStateMachine/tinyrelay/internal/gitrelay"
368+)
369+
370+func TestGitLiveEnabledRequiresGRASP02(t *testing.T) {
371+ _, tenant := testTenant(t)
372+ p := tenant.Policy()
373+ p.Features.Grasp = true
374+ p.Features.Grasp02 = false
375+ if err := tenant.applyPolicy(context.Background(), p); err != nil {
376+ t.Fatal(err)
377+ }
378+ if tenant.gitLiveEnabled() {
379+ t.Fatal("live synchronization enabled without GRASP-02")
380+ }
381+ p.Features.Grasp02 = true
382+ if err := tenant.applyPolicy(context.Background(), p); err != nil {
383+ t.Fatal(err)
384+ }
385+ if !tenant.gitLiveEnabled() {
386+ t.Fatal("live synchronization disabled with GRASP-02")
387+ }
388+}
389+
390+func TestGitLivePrivateTargetRequiresConfiguredPeer(t *testing.T) {
391+ _, tenant := testTenant(t)
392+ repo := gitrelay.Repository{Private: true}
393+ if tenant.gitLiveAuthorized(repo, "wss://peer.example") {
394+ t.Fatal("unconfigured private target was authorized")
395+ }
396+ p := tenant.Policy()
397+ p.Features.Grasp, p.Features.Grasp02, p.Features.Grasp08 = true, true, true
398+ p.Reads = "members"
399+ p.PrivatePeers = []string{"wss://peer.example"}
400+ if err := tenant.applyPolicy(context.Background(), p); err != nil {
401+ t.Fatal(err)
402+ }
403+ if !tenant.gitLiveAuthorized(repo, "wss://peer.example") {
404+ t.Fatal("configured private target was rejected")
405+ }
406+ if tenant.gitLiveAuthorized(repo, "wss://other.example") {
407+ t.Fatal("different private target was authorized")
408+ }
409+}
410+
411+func TestGitLiveFiltersIncludeConversationRootsAndReplies(t *testing.T) {
412+ repo := gitrelay.Repository{Owner: strings.Repeat("a", 64), Identifier: "repo"}
413+ root := event.Event{Kind: event.KIND_GIT_ISSUE, ID: strings.Repeat("b", 64)}
414+ filters := gitLiveFiltersForRoots(repo, []event.Event{root})
415+ if len(filters) != 5 {
416+ t.Fatalf("got %d filters, want metadata, conversation and two reply filters", len(filters))
417+ }
418+ for _, key := range []string{"e", "E"} {
419+ found := false
420+ for _, filter := range filters {
421+ if len(filter.Tags[key]) == 1 && filter.Tags[key][0] == root.ID {
422+ found = true
423+ }
424+ }
425+ if !found {
426+ t.Fatalf("missing reply filter for %s", key)
427+ }
428+ }
429+}
430+
431+func TestGitLiveCohortRespectsSocketCapAndRotates(t *testing.T) {
432+ keys := make([]string, maxGitLivePeers+7)
433+ for i := range keys {
434+ keys[i] = strings.Repeat("a", i+1)
435+ }
436+ first := gitLiveCohort(keys, 0)
437+ second := gitLiveCohort(keys, maxGitLivePeers)
438+ if len(first) != maxGitLivePeers || len(second) != maxGitLivePeers {
439+ t.Fatalf("cohort sizes = %d, %d; want %d", len(first), len(second), maxGitLivePeers)
440+ }
441+ _, firstHas := first[keys[maxGitLivePeers-1]]
442+ _, secondHas := second[keys[maxGitLivePeers-1]]
443+ if firstHas == secondHas {
444+ t.Fatal("live cohort did not rotate")
445+ }
446+}
447+
448+func TestRotateGitAuthorsUsesDurableOffset(t *testing.T) {
449+ authors := []string{"a", "b", "c"}
450+ got := rotateGitAuthors(authors, 1)
451+ if strings.Join(got, ",") != "b,c,a" {
452+ t.Fatalf("rotated authors = %v", got)
453+ }
454+}
455+
456+func TestStopGitLiveWorkersCancelsAndWaits(t *testing.T) {
457+ done := make(chan struct{})
458+ canceled := make(chan struct{})
459+ workers := map[string]gitLiveWorker{"peer": {
460+ cancel: func() { close(canceled); close(done) },
461+ done: done,
462+ }}
463+ finished := make(chan struct{})
464+ go func() {
465+ stopGitLiveWorkers(workers)
466+ close(finished)
467+ }()
468+ select {
469+ case <-canceled:
470+ case <-time.After(time.Second):
471+ t.Fatal("worker was not canceled")
472+ }
473+ select {
474+ case <-finished:
475+ case <-time.After(time.Second):
476+ t.Fatal("stop returned before worker completion")
477+ }
478+ if len(workers) != 0 {
479+ t.Fatal("stopped workers remain in the registry")
480+ }
481+}
482diff --git a/internal/daemon/git_outbox.go b/internal/daemon/git_outbox.go
483new file mode 100644
484index 0000000..58a4289
485--- /dev/null
486+++ b/internal/daemon/git_outbox.go
487@@ -0,0 +1,274 @@
488+package daemon
489+
490+import (
491+ "context"
492+ "errors"
493+ "net/url"
494+ "sort"
495+ "strconv"
496+ "strings"
497+ "time"
498+
499+ "github.com/FelineStateMachine/tinyrelay/internal/event"
500+ "github.com/FelineStateMachine/tinyrelay/internal/replication"
501+)
502+
503+const (
504+ gitOutboxAuthorLimit = 32
505+ gitOutboxRelayLimit = 24
506+)
507+
508+// gitOutboxOptions describes the bounded, read-only part of GRASP-03. Seeds
509+// are the repository's announced relays. Private repositories must also set
510+// PrivatePeers; discovered public relays are never used for private work.
511+type gitOutboxOptions struct {
512+ Transport replication.PullTransport
513+ Seeds []string
514+ Private bool
515+ PrivatePeers []string
516+ AuthorOffset int
517+ SeedOffset int
518+}
519+
520+// discoverGitOutboxes asks announced relays for the NIP-65 lists of authors
521+// who took part in a repository conversation. A failed author or seed is
522+// isolated so one unavailable relay cannot prevent the remaining discussion
523+// from being imported.
524+func discoverGitOutboxes(ctx context.Context, options gitOutboxOptions, authors []string) ([]string, error) {
525+ items, err := discoverGitOutboxEvents(ctx, options, authors)
526+ result := make([]string, 0, gitOutboxRelayLimit)
527+ seen := make(map[string]struct{}, gitOutboxRelayLimit)
528+ for _, item := range items {
529+ for _, relay := range gitOutboxRelayTags(item) {
530+ if !gitOutboxAllowedRelay(relay, options.Private, options.PrivatePeers) {
531+ continue
532+ }
533+ canonical := normalizeRelayGit(relay)
534+ if _, ok := seen[canonical]; ok {
535+ continue
536+ }
537+ seen[canonical] = struct{}{}
538+ result = append(result, canonical)
539+ if len(result) == gitOutboxRelayLimit {
540+ return result, err
541+ }
542+ }
543+ }
544+ return result, err
545+}
546+
547+// discoverGitOutboxEvents returns the newest valid authored list of each
548+// supported kind. Callers should ingest these events with OriginImport so the
549+// relay list becomes available after a restart and to normal replication.
550+func discoverGitOutboxEvents(ctx context.Context, options gitOutboxOptions, authors []string) ([]event.Event, error) {
551+ if options.Transport == nil {
552+ return nil, errors.New("GRASP-03: outbox transport is required")
553+ }
554+ authors = boundedGitAuthorsFrom(authors, options.AuthorOffset)
555+ seeds := boundedGitRelaysFrom(options.Seeds, options.Private, options.PrivatePeers, options.SeedOffset)
556+ if len(authors) == 0 || len(seeds) == 0 {
557+ return nil, nil
558+ }
559+ newest := make(map[string]event.Event, len(authors)*3)
560+ var queryErrors []error
561+ workCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
562+ defer cancel()
563+ stop := false
564+ for _, seed := range seeds {
565+ for _, author := range authors {
566+ for _, kind := range []int{event.KIND_PROFILE, 10002, 10317} {
567+ if err := workCtx.Err(); err != nil {
568+ queryErrors = append(queryErrors, err)
569+ stop = true
570+ break
571+ }
572+ items, err := options.Transport.Query(workCtx, seed, gitOutboxFilter(author, kind))
573+ if err != nil {
574+ if len(queryErrors) < 8 {
575+ queryErrors = append(queryErrors, err)
576+ }
577+ continue
578+ }
579+ for _, item := range items {
580+ if event.Validate(item) != nil || item.PubKey != author || item.Kind != kind {
581+ continue
582+ }
583+ key := author + ":" + strconv.Itoa(item.Kind)
584+ if current, ok := newest[key]; !ok || item.CreatedAt > current.CreatedAt || (item.CreatedAt == current.CreatedAt && item.ID < current.ID) {
585+ newest[key] = item
586+ }
587+ }
588+ }
589+ if stop {
590+ break
591+ }
592+ }
593+ if stop {
594+ break
595+ }
596+ }
597+ result := make([]event.Event, 0, len(newest))
598+ for _, item := range newest {
599+ result = append(result, item)
600+ }
601+ sort.Slice(result, func(i, j int) bool {
602+ if result[i].PubKey == result[j].PubKey {
603+ return result[i].Kind < result[j].Kind
604+ }
605+ return result[i].PubKey < result[j].PubKey
606+ })
607+ return result, errors.Join(queryErrors...)
608+}
609+
610+func gitOutboxFilter(author string, kind int) event.Filter {
611+ limit := 1
612+ return event.Filter{Authors: []string{author}, Kinds: []int{kind}, Tags: map[string][]string{}, Limit: &limit}
613+}
614+
615+// gitConversationAuthors returns the repository participants whose outboxes
616+// may contain replies. The caller supplies events already admitted to the
617+// repository conversation, so arbitrary relay authors cannot expand the
618+// discovery set.
619+func gitConversationAuthors(events []event.Event, owner string, maintainers []string) []string {
620+ authors := make([]string, 0, len(maintainers)+1+len(events))
621+ appendAuthor := func(author string) {
622+ if len(author) != 64 || strings.Trim(author, "0123456789abcdefABCDEF") != "" {
623+ return
624+ }
625+ for _, existing := range authors {
626+ if existing == author {
627+ return
628+ }
629+ }
630+ authors = append(authors, author)
631+ }
632+ appendAuthor(owner)
633+ for _, maintainer := range maintainers {
634+ appendAuthor(maintainer)
635+ }
636+ for _, item := range events {
637+ switch item.Kind {
638+ case event.KIND_GIT_PATCH, event.KIND_GIT_PR, event.KIND_GIT_PR_UPDATE, event.KIND_GIT_ISSUE, 1111, 1630, 1631, 1632, 1633:
639+ appendAuthor(item.PubKey)
640+ }
641+ }
642+ return authors
643+}
644+
645+func boundedGitAuthors(authors []string) []string {
646+ return boundedGitAuthorsFrom(authors, 0)
647+}
648+
649+func boundedGitAuthorsFrom(authors []string, offset int) []string {
650+ seen := make(map[string]struct{}, len(authors))
651+ result := make([]string, 0, minGitInt(len(authors), gitOutboxAuthorLimit))
652+ if offset < 0 {
653+ offset = 0
654+ }
655+ valid := 0
656+ for _, author := range authors {
657+ if len(author) != 64 || strings.Trim(author, "0123456789abcdefABCDEF") != "" {
658+ continue
659+ }
660+ if _, ok := seen[author]; ok {
661+ continue
662+ }
663+ seen[author] = struct{}{}
664+ if valid < offset {
665+ valid++
666+ continue
667+ }
668+ valid++
669+ result = append(result, author)
670+ if len(result) == gitOutboxAuthorLimit {
671+ break
672+ }
673+ }
674+ return result
675+}
676+
677+func boundedGitRelays(relays []string, private bool, peers []string) []string {
678+ return boundedGitRelaysFrom(relays, private, peers, 0)
679+}
680+
681+func boundedGitRelaysFrom(relays []string, private bool, peers []string, offset int) []string {
682+ seen := make(map[string]struct{}, len(relays))
683+ valid := make([]string, 0, len(relays))
684+ for _, relay := range relays {
685+ if !gitOutboxAllowedRelay(relay, private, peers) {
686+ continue
687+ }
688+ canonical := normalizeRelayGit(relay)
689+ if _, ok := seen[canonical]; ok {
690+ continue
691+ }
692+ seen[canonical] = struct{}{}
693+ valid = append(valid, canonical)
694+ }
695+ if len(valid) == 0 {
696+ return nil
697+ }
698+ start := offset % len(valid)
699+ if start < 0 {
700+ start = 0
701+ }
702+ rotated := append(append([]string(nil), valid[start:]...), valid[:start]...)
703+ result := rotated
704+ if len(result) > gitOutboxRelayLimit {
705+ result = result[:gitOutboxRelayLimit]
706+ }
707+ return result
708+}
709+
710+func gitOutboxRelayTags(item event.Event) []string {
711+ name := ""
712+ switch item.Kind {
713+ case 10002:
714+ name = "r"
715+ case 10317:
716+ name = "g"
717+ default:
718+ return nil
719+ }
720+ result := make([]string, 0)
721+ for _, tag := range item.Tags {
722+ if len(tag) < 2 || tag[0] != name {
723+ continue
724+ }
725+ if item.Kind == 10002 && len(tag) > 2 && strings.EqualFold(tag[2], "read") {
726+ continue
727+ }
728+ result = append(result, tag[1])
729+ }
730+ return result
731+}
732+
733+func gitOutboxAllowedRelay(raw string, private bool, peers []string) bool {
734+ u, err := url.Parse(strings.TrimSpace(raw))
735+ if err != nil || u.User != nil || u.Fragment != "" || u.Host == "" {
736+ return false
737+ }
738+ if u.Scheme == "http" {
739+ u.Scheme = "ws"
740+ } else if u.Scheme == "https" {
741+ u.Scheme = "wss"
742+ }
743+ if (u.Scheme != "ws" && u.Scheme != "wss") || !safeGitOutboxHost(u.Hostname()) {
744+ return false
745+ }
746+ if !private {
747+ return true
748+ }
749+ return privatePeerMatch(u.String(), peers)
750+}
751+
752+func safeGitOutboxHost(host string) bool {
753+ return host != "" && !strings.ContainsAny(host, " \t\r\n")
754+}
755+
756+func minGitInt(a, b int) int {
757+ if a < b {
758+ return a
759+ }
760+ return b
761+}
762diff --git a/internal/daemon/git_outbox_sync.go b/internal/daemon/git_outbox_sync.go
763new file mode 100644
764index 0000000..4251c63
765--- /dev/null
766+++ b/internal/daemon/git_outbox_sync.go
767@@ -0,0 +1,171 @@
768+package daemon
769+
770+import (
771+ "context"
772+ "errors"
773+ "fmt"
774+ "time"
775+
776+ "github.com/FelineStateMachine/tinyrelay/internal/event"
777+ "github.com/FelineStateMachine/tinyrelay/internal/gitrelay"
778+ "github.com/FelineStateMachine/tinyrelay/internal/replication"
779+ "github.com/FelineStateMachine/tinyrelay/internal/storage"
780+)
781+
782+const gitOutboxOffsetPrefix = "grasp03.outbox-offset."
783+
784+type tenantGitOutboxTransport struct {
785+ tenant *Tenant
786+ repo gitrelay.Repository
787+}
788+
789+func (t tenantGitOutboxTransport) Query(ctx context.Context, target string, filter event.Filter) ([]event.Event, error) {
790+ requestCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
791+ defer cancel()
792+ transport, err := t.tenant.gitTargetTransport(requestCtx, t.repo, target)
793+ if err != nil {
794+ return nil, err
795+ }
796+ return transport.Query(requestCtx, target, filter)
797+}
798+
799+// gitOutboxSync imports participant relay lists, then asks each discovered
800+// outbox for the repository conversation. List events are ingested first so
801+// the discovered routing remains durable after a restart.
802+func (t *Tenant) gitOutboxSync(ctx context.Context, repo gitrelay.Repository) error {
803+ if t == nil || t.store == nil {
804+ return errors.New("GRASP-03: tenant store is required")
805+ }
806+ events, err := t.gitConversationEvents(ctx, repo)
807+ if err != nil {
808+ return err
809+ }
810+ authors := gitConversationAuthors(events, repo.Owner, repo.Maintainers)
811+ offset := t.gitOutboxOffset(ctx, repo)
812+ selectedAuthors := boundedGitAuthorsFrom(authors, offset%maxGitInt(len(authors), 1))
813+ private := repo.Private || t.PrivateServiceEnabled()
814+ options := gitOutboxOptions{Transport: tenantGitOutboxTransport{tenant: t, repo: repo}, Seeds: repo.Relays, Private: private, PrivatePeers: t.Policy().PrivatePeers, AuthorOffset: 0, SeedOffset: offset}
815+ listEvents, err := t.gitLocalOutboxEvents(ctx, selectedAuthors)
816+ if err != nil {
817+ return err
818+ }
819+ discovered, discoverErr := discoverGitOutboxEvents(ctx, options, selectedAuthors)
820+ listEvents = mergeGitOutboxEvents(listEvents, discovered)
821+ var syncErrors []error
822+ if err := t.ingestGitOutboxEvents(ctx, discovered); err != nil {
823+ syncErrors = append(syncErrors, err)
824+ }
825+ if discoverErr != nil {
826+ syncErrors = append(syncErrors, discoverErr)
827+ }
828+ if len(authors) > 0 {
829+ if err := t.store.PutSetting(ctx, gitOutboxOffsetKey(repo), (offset+gitOutboxAuthorLimit)%len(authors)); err != nil {
830+ syncErrors = append(syncErrors, fmt.Errorf("GRASP-03: save outbox offset: %w", err))
831+ }
832+ }
833+ targets := gitOutboxRelays(listEvents, options.Private, options.PrivatePeers)
834+ targets = rotateGitRelays(targets, offset)
835+ for _, target := range targets {
836+ targetCtx, cancel := context.WithTimeout(ctx, 8*time.Second)
837+ if err := t.gitSyncPeer(targetCtx, repo, target); err != nil {
838+ syncErrors = append(syncErrors, fmt.Errorf("GRASP-03: sync %s: %w", target, err))
839+ }
840+ cancel()
841+ }
842+ return errors.Join(syncErrors...)
843+}
844+
845+func (t *Tenant) gitLocalOutboxEvents(ctx context.Context, authors []string) ([]event.Event, error) {
846+ result := make([]event.Event, 0, len(authors)*3)
847+ for _, author := range authors {
848+ for _, kind := range []int{event.KIND_PROFILE, 10002, 10317} {
849+ rows, err := t.store.Query(ctx, event.Filter{Authors: []string{author}, Kinds: []int{kind}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1})
850+ if err != nil {
851+ return nil, fmt.Errorf("GRASP-03: read cached outbox: %w", err)
852+ }
853+ if len(rows.Events) > 0 {
854+ result = append(result, rows.Events[0])
855+ }
856+ }
857+ }
858+ return result, nil
859+}
860+
861+func rotateGitRelays(relays []string, offset int) []string {
862+ if len(relays) == 0 {
863+ return nil
864+ }
865+ start := offset % len(relays)
866+ return append(append([]string(nil), relays[start:]...), relays[:start]...)
867+}
868+
869+func maxGitInt(a, b int) int {
870+ if a > b {
871+ return a
872+ }
873+ return b
874+}
875+
876+func mergeGitOutboxEvents(existing, incoming []event.Event) []event.Event {
877+ result := append([]event.Event(nil), existing...)
878+ for _, item := range incoming {
879+ found := false
880+ for i := range result {
881+ if result[i].PubKey != item.PubKey || result[i].Kind != item.Kind {
882+ continue
883+ }
884+ found = true
885+ if item.CreatedAt > result[i].CreatedAt || (item.CreatedAt == result[i].CreatedAt && item.ID < result[i].ID) {
886+ result[i] = item
887+ }
888+ break
889+ }
890+ if !found {
891+ result = append(result, item)
892+ }
893+ }
894+ return result
895+}
896+
897+func gitOutboxRelays(items []event.Event, private bool, peers []string) []string {
898+ result := make([]string, 0, gitOutboxRelayLimit)
899+ seen := make(map[string]struct{}, gitOutboxRelayLimit)
900+ for _, item := range items {
901+ for _, raw := range gitOutboxRelayTags(item) {
902+ if !gitOutboxAllowedRelay(raw, private, peers) {
903+ continue
904+ }
905+ canonical := normalizeRelayGit(raw)
906+ if _, ok := seen[canonical]; ok {
907+ continue
908+ }
909+ seen[canonical] = struct{}{}
910+ result = append(result, canonical)
911+ if len(result) == gitOutboxRelayLimit {
912+ return result
913+ }
914+ }
915+ }
916+ return result
917+}
918+
919+func (t *Tenant) ingestGitOutboxEvents(ctx context.Context, items []event.Event) error {
920+ for _, item := range items {
921+ if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) {
922+ return fmt.Errorf("GRASP-03: import outbox list: %w", err)
923+ }
924+ }
925+ return nil
926+}
927+
928+func (t *Tenant) gitOutboxOffset(ctx context.Context, repo gitrelay.Repository) int {
929+ var offset int
930+ if err := t.store.GetSetting(ctx, gitOutboxOffsetKey(repo), &offset); err != nil || offset < 0 {
931+ return 0
932+ }
933+ return offset
934+}
935+
936+func gitOutboxOffsetKey(repo gitrelay.Repository) string {
937+ return gitOutboxOffsetPrefix + repo.Owner + ":" + repo.Identifier
938+}
939diff --git a/internal/daemon/git_outbox_test.go b/internal/daemon/git_outbox_test.go
940new file mode 100644
941index 0000000..b0f56b0
942--- /dev/null
943+++ b/internal/daemon/git_outbox_test.go
944@@ -0,0 +1,173 @@
945+package daemon
946+
947+import (
948+ "context"
949+ "fmt"
950+ "testing"
951+
952+ "github.com/FelineStateMachine/tinyrelay/internal/event"
953+)
954+
955+type gitOutboxTransport struct {
956+ queries map[string][]event.Event
957+ errors map[string]error
958+}
959+
960+func (t gitOutboxTransport) Query(_ context.Context, target string, _ event.Filter) ([]event.Event, error) {
961+ if err := t.errors[target]; err != nil {
962+ return nil, err
963+ }
964+ if items := t.queries[target]; items != nil {
965+ return items, nil
966+ }
967+ return nil, nil
968+}
969+
970+func TestGitOutboxFilterIncludesProfilesRelayListsAndRepositoryRelayLists(t *testing.T) {
971+ key := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
972+ filter := gitOutboxFilter(key, 10002)
973+ if len(filter.Authors) != 1 || filter.Authors[0] != key {
974+ t.Fatalf("authors = %#v", filter.Authors)
975+ }
976+ if !containsGitKind(filter.Kinds, 10002) || len(filter.Kinds) != 1 {
977+ t.Fatalf("kinds = %#v", filter.Kinds)
978+ }
979+ if filter.Limit == nil || *filter.Limit != 1 {
980+ t.Fatalf("limit = %#v", filter.Limit)
981+ }
982+}
983+
984+func TestGitOutboxAuthorsAndRelaysAreDeduplicatedAndBounded(t *testing.T) {
985+ key := "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
986+ authors := make([]string, 0, gitOutboxAuthorLimit+2)
987+ for i := 0; i < gitOutboxAuthorLimit+2; i++ {
988+ authors = append(authors, key)
989+ }
990+ if got := boundedGitAuthors(authors); len(got) != 1 {
991+ t.Fatalf("authors = %#v", got)
992+ }
993+ relays := boundedGitRelays([]string{"https://Relay.EXAMPLE/", "wss://relay.example"}, false, nil)
994+ if len(relays) != 1 || relays[0] != "wss://relay.example" {
995+ t.Fatalf("relays = %#v", relays)
996+ }
997+}
998+
999+func TestGitOutboxPrivateDiscoveryNeverUsesUnconfiguredRelay(t *testing.T) {
1000+ if gitOutboxAllowedRelay("wss://other.example", true, []string{"https://peer.example/git"}) {
1001+ t.Fatal("unconfigured private relay was allowed")
1002+ }
1003+ if !gitOutboxAllowedRelay("wss://peer.example/git", true, []string{"https://peer.example/git"}) {
1004+ t.Fatal("configured private relay was rejected")
1005+ }
1006+}
1007+
1008+func TestGitOutboxRelayTagsUseWriteCapableNIP65Entries(t *testing.T) {
1009+ list := event.Event{Kind: 10002, Tags: [][]string{{"r", "wss://read.example", "read"}, {"r", "wss://write.example", "write"}, {"r", "wss://both.example"}}}
1010+ got := gitOutboxRelayTags(list)
1011+ if len(got) != 2 || got[0] != "wss://write.example" || got[1] != "wss://both.example" {
1012+ t.Fatalf("relays = %#v", got)
1013+ }
1014+}
1015+
1016+func TestGitOutboxRelayTagsUseGroupEntriesForKind10317(t *testing.T) {
1017+ list := event.Event{Kind: 10317, Tags: [][]string{{"r", "wss://wrong.example"}, {"g", "wss://group.example"}}}
1018+ got := gitOutboxRelayTags(list)
1019+ if len(got) != 1 || got[0] != "wss://group.example" {
1020+ t.Fatalf("relays = %#v", got)
1021+ }
1022+}
1023+
1024+func TestGitConversationAuthorsAreLimitedToAdmittedParticipants(t *testing.T) {
1025+ owner := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
1026+ maintainer := "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
1027+ participant := "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
1028+ authors := gitConversationAuthors([]event.Event{{Kind: event.KIND_GIT_ISSUE, PubKey: participant}, {Kind: event.KIND_PROFILE, PubKey: "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"}}, owner, []string{maintainer})
1029+ if len(authors) != 3 || authors[0] != owner || authors[1] != maintainer || authors[2] != participant {
1030+ t.Fatalf("authors = %#v", authors)
1031+ }
1032+}
1033+
1034+func TestGitConversationAuthorsDoesNotDropParticipantsBeforeBatching(t *testing.T) {
1035+ events := make([]event.Event, 40)
1036+ for i := range events {
1037+ events[i] = event.Event{Kind: event.KIND_GIT_ISSUE, PubKey: fmt.Sprintf("%064x", i+1)}
1038+ }
1039+ if got := gitConversationAuthors(events, "", nil); len(got) != len(events) {
1040+ t.Fatalf("authors = %d, want %d", len(got), len(events))
1041+ }
1042+}
1043+
1044+func TestGitOutboxAuthorBatchRotatesPastLimit(t *testing.T) {
1045+ events := make([]event.Event, 40)
1046+ for i := range events {
1047+ events[i] = event.Event{Kind: event.KIND_GIT_ISSUE, PubKey: fmt.Sprintf("%064x", i+1)}
1048+ }
1049+ authors := gitConversationAuthors(events, "", nil)
1050+ first := boundedGitAuthorsFrom(authors, 0)
1051+ second := boundedGitAuthorsFrom(authors, gitOutboxAuthorLimit)
1052+ if len(first) != gitOutboxAuthorLimit || len(second) != len(authors)-gitOutboxAuthorLimit {
1053+ t.Fatalf("batch sizes = %d and %d", len(first), len(second))
1054+ }
1055+ if first[0] == second[0] {
1056+ t.Fatal("rotated batch repeated its first author")
1057+ }
1058+}
1059+
1060+func TestRotateGitRelaysChangesStartingPoint(t *testing.T) {
1061+ input := []string{"wss://one.example", "wss://two.example", "wss://three.example"}
1062+ got := rotateGitRelays(input, 1)
1063+ if len(got) != 3 || got[0] != input[1] || got[2] != input[0] {
1064+ t.Fatalf("rotated relays = %#v", got)
1065+ }
1066+ if input[0] != "wss://one.example" {
1067+ t.Fatal("rotation modified input")
1068+ }
1069+}
1070+
1071+func TestDiscoverGitOutboxesRequiresTransport(t *testing.T) {
1072+ _, err := discoverGitOutboxes(context.Background(), gitOutboxOptions{}, []string{"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"})
1073+ if err == nil || err.Error() != "GRASP-03: outbox transport is required" {
1074+ t.Fatalf("error = %v", err)
1075+ }
1076+}
1077+
1078+func TestDiscoverGitOutboxesIsolatesUnavailableSeeds(t *testing.T) {
1079+ secret, err := event.GenerateKey()
1080+ if err != nil {
1081+ t.Fatal(err)
1082+ }
1083+ item := event.Event{Kind: 10002, CreatedAt: 1, Tags: [][]string{{"r", "wss://outbox.example"}}}
1084+ if err := event.Sign(&item, secret); err != nil {
1085+ t.Fatal(err)
1086+ }
1087+ if err := event.Validate(item); err != nil {
1088+ t.Fatal(err)
1089+ }
1090+ if got := gitOutboxRelayTags(item); len(got) != 1 {
1091+ t.Fatalf("item relays = %#v", got)
1092+ }
1093+ transport := gitOutboxTransport{
1094+ queries: map[string][]event.Event{"wss://healthy.example": {item}},
1095+ errors: map[string]error{"wss://down.example": fmt.Errorf("connection refused")},
1096+ }
1097+ authors := []string{item.PubKey}
1098+ if got := boundedGitRelays([]string{"wss://down.example", "wss://healthy.example"}, false, nil); len(got) != 2 {
1099+ t.Fatalf("seed relays = %#v", got)
1100+ }
1101+ got, err := discoverGitOutboxes(context.Background(), gitOutboxOptions{Transport: transport, Seeds: []string{"wss://down.example", "wss://healthy.example"}}, authors)
1102+ if err == nil {
1103+ t.Fatal("expected isolated seed error to be reported")
1104+ }
1105+ if len(got) != 1 || got[0] != "wss://outbox.example" {
1106+ t.Fatalf("relays = %#v, err = %v", got, err)
1107+ }
1108+}
1109+
1110+func containsGitKind(kinds []int, wanted int) bool {
1111+ for _, kind := range kinds {
1112+ if kind == wanted {
1113+ return true
1114+ }
1115+ }
1116+ return false
1117+}
1118diff --git a/internal/daemon/git_services.go b/internal/daemon/git_services.go
1119index 32a9746..51e0a40 100644
1120--- a/internal/daemon/git_services.go
1121+++ b/internal/daemon/git_services.go
1122@@ -6,83 +6,231 @@ import (
1123 "net/url"
1124 "sort"
1125 "strings"
1126+ "time"
1127
1128 "github.com/FelineStateMachine/tinyrelay/internal/event"
1129 "github.com/FelineStateMachine/tinyrelay/internal/gitrelay"
1130 "github.com/FelineStateMachine/tinyrelay/internal/replication"
1131 "github.com/FelineStateMachine/tinyrelay/internal/storage"
1132+ "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"
1133 )
1134
1135-// gitSync is the GRASP-02 callback passed to GitRelay. It repairs every
1136-// signed repository tip from the announcement's clone/relay sources before
1137-// allowing the Git relay to expose the state.
1138+// gitSync repairs signed branch state. Conversation history is scheduled
1139+// separately so a missing Git object cannot block issue or PR replication.
1140 func (t *Tenant) gitSync(ctx context.Context, repo gitrelay.Repository) error {
1141 if len(repo.Refs) == 0 {
1142 return nil
1143 }
1144- if len(gitrelay.RepairSources(repo)) == 0 {
1145- return errors.New("GRASP-02: repository has no repair source")
1146- }
1147+ repo.Private = repo.Private || t.PrivateServiceEnabled()
1148 t.git.ConfigurePrivatePeers(t.Policy().PrivatePeers, t.privateHTTPAuth)
1149 return t.git.FetchMissing(ctx, repo, gitrelay.RepairSources(repo), repo.Refs)
1150 }
1151
1152-// gitEventSync is the GRASP-03 callback. It imports NIP-34 repository
1153-// metadata from announced relays through the hardened replication transport;
1154-// ordinary event policy and GitRelay admission remain the final gates.
1155 func (t *Tenant) gitEventSync(ctx context.Context, repo gitrelay.Repository) error {
1156- if len(repo.Relays) == 0 {
1157- return nil
1158+ var result error
1159+ targets := boundedGitRelays(repo.Relays, repo.Private || t.PrivateServiceEnabled(), t.Policy().PrivatePeers)
1160+ setting := "grasp02.peer-offset." + repo.Owner + ":" + repo.Identifier
1161+ var offset int
1162+ _ = t.store.GetSetting(ctx, setting, &offset)
1163+ if offset < 0 {
1164+ offset = 0
1165 }
1166- filters := gitEventFilters(repo)
1167- for _, target := range repo.Relays {
1168+ targets = rotateGitRelays(targets, offset)
1169+ historyCtx, historyCancel := context.WithTimeout(ctx, 20*time.Second)
1170+ for index, target := range targets {
1171+ if historyCtx.Err() != nil {
1172+ break
1173+ }
1174 if sameRelayGit(target, t.RelayURL(), t.publicURL) {
1175 continue
1176 }
1177- transport := &replication.NostrTransport{Dialer: t.gitEventDialer(ctx, target)}
1178- seen := make(map[string]struct{})
1179- roots := make([]string, 0, 16)
1180- for _, filter := range filters {
1181- items, err := transport.Query(ctx, target, filter)
1182- if err != nil {
1183- return err
1184- }
1185- sort.SliceStable(items, func(i, j int) bool {
1186- return metadataEventOrder(items[i].Kind) < metadataEventOrder(items[j].Kind)
1187- })
1188- for _, item := range items {
1189- if _, ok := seen[item.ID]; ok {
1190- continue
1191- }
1192- seen[item.ID] = struct{}{}
1193- if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE {
1194- roots = append(roots, item.ID)
1195- }
1196- if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) {
1197- return err
1198- }
1199+ // Persist the next starting peer before a bounded network attempt.
1200+ // An unavailable peer cannot monopolize every subsequent pass.
1201+ result = errors.Join(result, t.store.PutSetting(ctx, setting, (offset+index+1)%len(targets)))
1202+ peerCtx, cancel := context.WithTimeout(historyCtx, 8*time.Second)
1203+ result = errors.Join(result, t.gitSyncPeer(peerCtx, repo, target))
1204+ cancel()
1205+ }
1206+ historyCancel()
1207+ repairCtx, repairCancel := context.WithTimeout(ctx, 5*time.Second)
1208+ result = errors.Join(result, t.repairGitPullRequests(repairCtx, repo))
1209+ repairCancel()
1210+ if t.Policy().Features.Grasp03 {
1211+ outboxCtx, outboxCancel := context.WithTimeout(ctx, 15*time.Second)
1212+ result = errors.Join(result, t.gitOutboxSync(outboxCtx, repo))
1213+ outboxCancel()
1214+ }
1215+ return withoutBudgetDeadline(ctx, result)
1216+}
1217+
1218+// withoutBudgetDeadline drops deadlines from the pass's own time budgets.
1219+// Peer offsets and outbox offsets persist between passes, so running out of
1220+// time leaves the repository incomplete rather than failed and keeps it on the
1221+// regular cadence. Cancellation of the caller's context still propagates.
1222+func withoutBudgetDeadline(parent context.Context, err error) error {
1223+ if err == nil || parent.Err() != nil {
1224+ return err
1225+ }
1226+ if joined, ok := err.(interface{ Unwrap() []error }); ok {
1227+ var kept []error
1228+ for _, item := range joined.Unwrap() {
1229+ if item = withoutBudgetDeadline(parent, item); item != nil {
1230+ kept = append(kept, item)
1231 }
1232 }
1233+ return errors.Join(kept...)
1234+ }
1235+ if errors.Is(err, context.DeadlineExceeded) {
1236+ return nil
1237+ }
1238+ return err
1239+}
1240+
1241+// gitTargetTransport checks the privacy boundary before sending any filters.
1242+// An unverified private peer never falls back to an anonymous connection.
1243+func (t *Tenant) gitTargetTransport(ctx context.Context, repo gitrelay.Repository, target string) (*replication.NostrTransport, error) {
1244+ p := t.Policy()
1245+ if !p.Features.Grasp || !p.Features.Grasp02 {
1246+ return nil, errors.New("GRASP synchronization is disabled")
1247+ }
1248+ if !gitOutboxAllowedRelay(target, repo.Private || t.PrivateServiceEnabled(), p.PrivatePeers) {
1249+ return nil, errors.New("GRASP synchronization target is not allowed")
1250+ }
1251+ if repo.Private || t.PrivateServiceEnabled() {
1252+ if !t.PrivateServiceEnabled() || !privatePeerMatch(target, p.PrivatePeers) || !t.privatePeerReady(ctx, target) {
1253+ return nil, errors.New("private peer is not ready")
1254+ }
1255+ }
1256+ return &replication.NostrTransport{Dialer: t.gitEventDialer(repo.Private || t.PrivateServiceEnabled()), Timeout: 5 * time.Second}, nil
1257+}
1258+
1259+func (t *Tenant) gitSyncPeer(ctx context.Context, repo gitrelay.Repository, target string) error {
1260+ transport, err := t.gitTargetTransport(ctx, repo, target)
1261+ if err != nil {
1262+ return err
1263+ }
1264+ var result error
1265+ for _, filter := range gitEventFilters(repo) {
1266+ result = errors.Join(result, t.gitPullFilter(ctx, transport, target, filter))
1267+ }
1268+ events, err := t.gitConversationEvents(ctx, repo)
1269+ if err != nil {
1270+ return errors.Join(result, err)
1271+ }
1272+ for _, filter := range gitReplyFilters(events) {
1273+ result = errors.Join(result, t.gitPullFilter(ctx, transport, target, filter))
1274+ }
1275+ return result
1276+}
1277+
1278+func (t *Tenant) gitPullFilter(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) error {
1279+ local, err := t.localSyncItems(ctx, filter)
1280+ if err != nil {
1281+ return err
1282+ }
1283+ items, pullErr := transport.QuerySynchronized(ctx, target, filter, local)
1284+ sort.SliceStable(items, func(i, j int) bool { return metadataEventOrder(items[i].Kind) < metadataEventOrder(items[j].Kind) })
1285+ for _, item := range items {
1286+ if ctx.Err() != nil {
1287+ return errors.Join(pullErr, ctx.Err())
1288+ }
1289+ if !event.Matches(filter, item) || event.Validate(item) != nil {
1290+ continue
1291+ }
1292+ // Rejected remote events must not prevent later valid events from arriving.
1293+ if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) {
1294+ t.app.telemetry.Logger().Debug("Git sync event rejected", "kind", item.Kind, "error", err)
1295+ }
1296+ }
1297+ return pullErr
1298+}
1299+
1300+func (t *Tenant) localSyncItems(ctx context.Context, filter event.Filter) ([]syncprotocol.Item, error) {
1301+ rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10001})
1302+ if err != nil {
1303+ return nil, err
1304+ }
1305+ if rows.More {
1306+ return nil, replication.ErrPullIncomplete
1307+ }
1308+ items := make([]syncprotocol.Item, 0, len(rows.Events))
1309+ for _, item := range rows.Events {
1310+ items = append(items, syncprotocol.Item{ID: item.ID, Timestamp: item.CreatedAt})
1311+ }
1312+ return items, nil
1313+}
1314+
1315+func (t *Tenant) gitConversationEvents(ctx context.Context, repo gitrelay.Repository) ([]event.Event, error) {
1316+ var items []event.Event
1317+ for _, filter := range gitEventFilters(repo)[1:] {
1318+ rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10000})
1319+ if err != nil {
1320+ return nil, err
1321+ }
1322+ if rows.More {
1323+ return nil, replication.ErrPullIncomplete
1324+ }
1325+ items = append(items, rows.Events...)
1326+ }
1327+ roots := uniqueEvents(items)
1328+ for _, filter := range gitReplyFilters(roots) {
1329+ rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10000})
1330+ if err != nil {
1331+ return nil, err
1332+ }
1333+ if rows.More {
1334+ return nil, replication.ErrPullIncomplete
1335+ }
1336+ items = append(items, rows.Events...)
1337+ }
1338+ return uniqueEvents(items), nil
1339+}
1340+
1341+func gitReplyFilters(items []event.Event) []event.Filter {
1342+ var ids []string
1343+ for _, item := range items {
1344+ if item.Kind == 1617 || item.Kind == 1618 || item.Kind == 1621 {
1345+ ids = append(ids, item.ID)
1346+ }
1347+ }
1348+ var filters []event.Filter
1349+ for start := 0; start < len(ids); start += 100 {
1350+ end := start + 100
1351+ if end > len(ids) {
1352+ end = len(ids)
1353+ }
1354 for _, key := range []string{"e", "E"} {
1355- if len(roots) == 0 {
1356- break
1357- }
1358- items, err := transport.Query(ctx, target, event.Filter{Kinds: []int{event.KIND_GIT_PR_UPDATE, 1111, 1630, 1631, 1632, 1633}, Tags: map[string][]string{key: roots}})
1359- if err != nil {
1360- return err
1361- }
1362- for _, item := range items {
1363- if _, ok := seen[item.ID]; ok {
1364- continue
1365- }
1366- seen[item.ID] = struct{}{}
1367- if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) {
1368- return err
1369- }
1370+ filters = append(filters, event.Filter{Kinds: []int{1619, 1111, 1630, 1631, 1632, 1633}, Tags: map[string][]string{key: ids[start:end]}})
1371+ }
1372+ }
1373+ return filters
1374+}
1375+
1376+func (t *Tenant) repairGitPullRequests(ctx context.Context, repo gitrelay.Repository) error {
1377+ repo.Private = repo.Private || t.PrivateServiceEnabled()
1378+ items, err := t.gitConversationEvents(ctx, repo)
1379+ if err != nil {
1380+ return err
1381+ }
1382+ t.git.ConfigurePrivatePeers(t.Policy().PrivatePeers, t.privateHTTPAuth)
1383+ var result error
1384+ for _, root := range items {
1385+ if root.Kind != 1618 {
1386+ continue
1387+ }
1388+ set := gitrelay.PullRequestRepair{Root: root}
1389+ for _, update := range items {
1390+ if update.Kind == 1619 && update.PubKey == root.PubKey && event.Tag(update, "E") == root.ID && event.Tag(update, "P") == root.PubKey {
1391+ set.Updates = append(set.Updates, update)
1392 }
1393 }
1394+ err := t.git.RepairPullRequestObjects(ctx, repo, set)
1395+ result = errors.Join(result, err)
1396+ if ctx.Err() != nil {
1397+ return errors.Join(result, ctx.Err())
1398+ }
1399 }
1400- return nil
1401+ return result
1402 }
1403
1404 func metadataEventOrder(kind int) int {
1405@@ -92,9 +240,9 @@ func metadataEventOrder(kind int) int {
1406 return 1
1407 }
1408
1409-func (t *Tenant) gitEventDialer(ctx context.Context, target string) replication.WebsocketDialer {
1410+func (t *Tenant) gitEventDialer(private bool) replication.WebsocketDialer {
1411 dialer := replication.WebsocketDialer{AllowPrivate: t.app.cfg.AllowPrivateRelays, MaxMessageBytes: t.app.cfg.MaxMessageBytes}
1412- if !t.privatePeerReady(ctx, target) {
1413+ if !private {
1414 return dialer
1415 }
1416 dialer.Authenticate = func(authCtx context.Context, relayURL, challenge string) (event.Event, error) {
1417@@ -135,5 +283,29 @@ func normalizeRelayGit(raw string) string {
1418 case "https":
1419 u.Scheme = "wss"
1420 }
1421+ u.Host = strings.ToLower(u.Host)
1422 return strings.TrimRight(u.String(), "/")
1423 }
1424+
1425+func uniqueEvents(rows []event.Event) []event.Event {
1426+ seen := make(map[string]struct{}, len(rows))
1427+ result := make([]event.Event, 0, len(rows))
1428+ for _, row := range rows {
1429+ if _, ok := seen[row.ID]; ok {
1430+ continue
1431+ }
1432+ seen[row.ID] = struct{}{}
1433+ result = append(result, row)
1434+ }
1435+ return result
1436+}
1437+
1438+// sortEventsNewestFirst matches the stable storage order used by cursors.
1439+func sortEventsNewestFirst(rows []event.Event) {
1440+ sort.Slice(rows, func(i, j int) bool {
1441+ if rows[i].CreatedAt == rows[j].CreatedAt {
1442+ return rows[i].ID < rows[j].ID
1443+ }
1444+ return rows[i].CreatedAt > rows[j].CreatedAt
1445+ })
1446+}
1447diff --git a/internal/daemon/git_sync_integration_test.go b/internal/daemon/git_sync_integration_test.go
1448new file mode 100644
1449index 0000000..8bfae17
1450--- /dev/null
1451+++ b/internal/daemon/git_sync_integration_test.go
1452@@ -0,0 +1,25 @@
1453+package daemon
1454+
1455+import (
1456+ "context"
1457+ "testing"
1458+
1459+ "github.com/FelineStateMachine/tinyrelay/internal/gitrelay"
1460+)
1461+
1462+func TestGitTargetTransportRequiresFeatureAndPrivatePeer(t *testing.T) {
1463+ _, tenant := testTenant(t)
1464+ repo := gitrelay.Repository{Owner: tenant.Policy().Owner, Identifier: "test"}
1465+ if _, err := tenant.gitTargetTransport(context.Background(), repo, "wss://unconfigured.example"); err == nil {
1466+ t.Fatal("disabled sync opened a transport")
1467+ }
1468+ p := tenant.Policy()
1469+ p.Features.Grasp, p.Features.Grasp02, p.Features.Grasp08 = true, true, true
1470+ p.Reads = "members"
1471+ if err := tenant.applyPolicy(context.Background(), p); err != nil {
1472+ t.Fatal(err)
1473+ }
1474+ if _, err := tenant.gitTargetTransport(context.Background(), repo, "wss://unconfigured.example"); err == nil {
1475+ t.Fatal("private repository could send filters to an unconfigured peer")
1476+ }
1477+}
1478diff --git a/internal/daemon/scheduler.go b/internal/daemon/scheduler.go
1479index 6d9174d..2722aa4 100644
1480--- a/internal/daemon/scheduler.go
1481+++ b/internal/daemon/scheduler.go
1482@@ -50,7 +50,6 @@ func (t *Tenant) runScheduler() {
1483 }
1484 report("retention", t.sweep(ctx, now.Unix()))
1485 report("restore-state", t.sweepReplicationState(ctx, now.Unix()))
1486- report("git-sync", t.git.GRASPService().Tick(ctx))
1487 nextMaintenance = now.Add(time.Minute)
1488 }
1489 next = nextMaintenance
1490diff --git a/internal/daemon/services.go b/internal/daemon/services.go
1491index b31b3b7..a3aea8d 100644
1492--- a/internal/daemon/services.go
1493+++ b/internal/daemon/services.go
1494@@ -137,6 +137,10 @@ func (t *Tenant) initServices(ctx context.Context) error {
1495 }()
1496 t.workWG.Add(1)
1497 go func() { defer t.workWG.Done(); t.runScheduler() }()
1498+ t.workWG.Add(1)
1499+ go func() { defer t.workWG.Done(); t.runGitLive(t.workCtx) }()
1500+ t.workWG.Add(1)
1501+ go func() { defer t.workWG.Done(); t.runGitHistory(t.workCtx) }()
1502 return nil
1503 }
1504
1505diff --git a/internal/gitrelay/gitrelay.go b/internal/gitrelay/gitrelay.go
1506index e15c2e1..60ef1f6 100644
1507--- a/internal/gitrelay/gitrelay.go
1508+++ b/internal/gitrelay/gitrelay.go
1509@@ -150,7 +150,7 @@ func (g *GitRelay) ValidateImported(ctx context.Context, e event.Event) (Reposit
1510
1511 const (
1512 graspRetryDelay = int64(5 * 60)
1513- graspSuccessDelay = int64(60 * 60)
1514+ graspSuccessDelay = int64(55 * 60)
1515 )
1516
1517 type tickProgress struct {
1518@@ -236,54 +236,60 @@ func (s *Service) Tick(ctx context.Context) error {
1519 var last string
1520 progress := tickProgress{At: time.Now().Unix(), Repos: make(map[string]tickRepository, len(repos))}
1521 previousProgress, _ := s.loadProgress()
1522+ // Resume after the last attempted repository if the prior pass ran out
1523+ // of time. Preserve untouched statuses instead of marking them failed.
1524+ last = previousProgress.LastKey
1525+ for _, r := range repos {
1526+ if previous, ok := previousProgress.Repos[key(r.Owner, r.Identifier)]; ok {
1527+ progress.Repos[key(r.Owner, r.Identifier)] = previous
1528+ }
1529+ }
1530+ start := sort.Search(len(repos), func(i int) bool { return key(repos[i].Owner, repos[i].Identifier) > last })
1531+ repos = append(repos[start:], repos[:start]...)
1532 var tickErr error
1533 for _, r := range repos {
1534- last = key(r.Owner, r.Identifier)
1535+ if ctx.Err() != nil {
1536+ tickErr = errors.Join(tickErr, ctx.Err())
1537+ break
1538+ }
1539+ repoKey := key(r.Owner, r.Identifier)
1540 status := tickRepository{AttemptedAt: progress.At, Attempts: 1}
1541 profile := s.relay.policy()
1542 status.Fingerprint = repoFingerprint(r, profile)
1543- if previous, ok := previousProgress.Repos[last]; ok {
1544+ if previous, ok := previousProgress.Repos[repoKey]; ok {
1545 if previous.NextAt > progress.At && previous.Fingerprint == status.Fingerprint {
1546- progress.Repos[last] = previous
1547+ progress.Repos[repoKey] = previous
1548 continue
1549 }
1550 if previous.Error != "" {
1551 status.Attempts = previous.Attempts + 1
1552 }
1553 }
1554+ last = repoKey
1555 features := profile.Features
1556- if features.Grasp02 && s.relay.gitSync != nil {
1557- if err := s.relay.gitSync(ctx, r); err != nil {
1558- status.Error = err.Error()
1559- status.NextAt = progress.At + graspRetryDelay
1560- tickErr = errors.Join(tickErr, fmt.Errorf("%s Git sync: %w", last, err))
1561- progress.Repos[last] = status
1562- continue
1563+ var repoErr error
1564+ if features.Grasp && features.Grasp02 {
1565+ if s.relay.eventSync != nil {
1566+ eventCtx, cancel := context.WithTimeout(ctx, 40*time.Second)
1567+ repoErr = errors.Join(repoErr, s.relay.eventSync(eventCtx, r))
1568+ cancel()
1569 }
1570- } else if features.Grasp02 && len(r.Refs) > 0 && len(RepairSources(r)) > 0 {
1571- // Native fallback for self-hosted deployments: fetch and verify the
1572- // signed tips without requiring a hosted-worker callback.
1573- if err := s.relay.FetchMissing(ctx, r, RepairSources(r), r.Refs); err != nil {
1574- // A source may be temporarily unavailable; leave the persisted
1575- // cursor intact and let the next tick retry.
1576- status.Error = err.Error()
1577- status.NextAt = progress.At + graspRetryDelay
1578- tickErr = errors.Join(tickErr, fmt.Errorf("%s Git sync: %w", last, err))
1579- progress.Repos[last] = status
1580- continue
1581+ gitCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
1582+ if s.relay.gitSync != nil {
1583+ repoErr = errors.Join(repoErr, s.relay.gitSync(gitCtx, r))
1584+ } else if len(r.Refs) > 0 && len(RepairSources(r)) > 0 {
1585+ repoErr = errors.Join(repoErr, s.relay.FetchMissing(gitCtx, r, RepairSources(r), r.Refs))
1586 }
1587+ cancel()
1588 }
1589- if features.Grasp03 && s.relay.eventSync != nil {
1590- if err := s.relay.eventSync(ctx, r); err != nil {
1591- status.Error = err.Error()
1592- status.NextAt = progress.At + graspRetryDelay
1593- tickErr = errors.Join(tickErr, fmt.Errorf("%s event sync: %w", last, err))
1594- progress.Repos[last] = status
1595- continue
1596- }
1597+ if repoErr != nil {
1598+ status.Error = repoErr.Error()
1599+ status.NextAt = progress.At + graspRetryDelay
1600+ tickErr = errors.Join(tickErr, fmt.Errorf("%s synchronization: %w", last, repoErr))
1601+ } else {
1602+ status.SucceededAt = progress.At
1603+ status.NextAt = progress.At + graspSuccessDelay
1604 }
1605- status.SucceededAt = progress.At
1606- status.NextAt = progress.At + graspSuccessDelay
1607 progress.Repos[last] = status
1608 }
1609 progress.LastKey = last
1610diff --git a/internal/gitrelay/pr_repair.go b/internal/gitrelay/pr_repair.go
1611new file mode 100644
1612index 0000000..7403d68
1613--- /dev/null
1614+++ b/internal/gitrelay/pr_repair.go
1615@@ -0,0 +1,138 @@
1616+package gitrelay
1617+
1618+import (
1619+ "context"
1620+ "errors"
1621+ "fmt"
1622+
1623+ "github.com/FelineStateMachine/tinyrelay/internal/event"
1624+)
1625+
1626+// PullRequestRepair groups a PR with its later tip updates. The event store
1627+// remains the authority for these events; this value only describes the Git
1628+// objects that GRASP-02 should fetch for them.
1629+type PullRequestRepair struct {
1630+ Root event.Event
1631+ Updates []event.Event
1632+}
1633+
1634+// RepairPullRequestObjects fetches the commits named by a NIP-34 PR and its
1635+// updates into the supplied hosted repository.
1636+//
1637+// Updates are accepted only from the PR author and must point to the root PR.
1638+// This prevents an unrelated event from supplying a ref to a hosted PR.
1639+func (g *GitRelay) RepairPullRequestObjects(ctx context.Context, hosted Repository, set PullRequestRepair) error {
1640+ root, err := pullRequestTarget(set.Root, event.KIND_GIT_PR)
1641+ if err != nil {
1642+ return err
1643+ }
1644+ if !hasRepositoryCoordinate(set.Root, hosted) {
1645+ return errors.New("blocked: pull request targets another repository")
1646+ }
1647+ // The hosted repository owns its path and access policy. PR events only
1648+ // contribute temporary refs and clone sources.
1649+ root = hosted
1650+ root.Alternative = false
1651+ sources := cloneURLs(set.Root)
1652+ refs := map[string]string{"refs/nostr/" + set.Root.ID: event.Tag(set.Root, "c")}
1653+ for _, update := range set.Updates {
1654+ if err := validatePullRequestUpdate(set.Root, update); err != nil {
1655+ return err
1656+ }
1657+ commit := event.Tag(update, "c")
1658+ if commit == "" {
1659+ continue
1660+ }
1661+ refs["refs/nostr/"+update.ID] = commit
1662+ sources = append(sources, cloneURLs(update)...)
1663+ }
1664+ if len(sources) == 0 {
1665+ return errors.New("GRASP-02: pull request has no clone source")
1666+ }
1667+ sources = uniqueStrings(sources)
1668+ if err := g.FetchMissing(ctx, root, sources, refs); err != nil {
1669+ return fmt.Errorf("GRASP-02 pull request repair: %w", err)
1670+ }
1671+ return nil
1672+}
1673+
1674+func hasRepositoryCoordinate(pr event.Event, repo Repository) bool {
1675+ want := "30617:" + repo.Owner + ":" + repo.Identifier
1676+ for _, tag := range pr.Tags {
1677+ if len(tag) > 1 && (tag[0] == "a" || tag[0] == "A") && tag[1] == want {
1678+ return true
1679+ }
1680+ }
1681+ return false
1682+}
1683+
1684+func cloneURLs(e event.Event) []string {
1685+ var out []string
1686+ for _, tag := range e.Tags {
1687+ if len(tag) > 1 && tag[0] == "clone" {
1688+ out = append(out, tag[1:]...)
1689+ }
1690+ }
1691+ return out
1692+}
1693+
1694+func pullRequestTarget(raw event.Event, kind int) (Repository, error) {
1695+ if err := event.Validate(raw); err != nil {
1696+ return Repository{}, err
1697+ }
1698+ if raw.Kind != kind {
1699+ return Repository{}, fmt.Errorf("unsupported: event kind %d", raw.Kind)
1700+ }
1701+ commit := event.Tag(raw, "c")
1702+ if !isObjectID(commit) {
1703+ return Repository{}, errors.New("invalid: pull request commit is not a SHA-1")
1704+ }
1705+ return Repository{
1706+ Owner: raw.PubKey,
1707+ Identifier: raw.ID,
1708+ EventID: raw.ID,
1709+ Alternative: true,
1710+ Clone: cloneURLs(raw),
1711+ Refs: map[string]string{"refs/nostr/" + raw.ID: commit},
1712+ }, nil
1713+}
1714+
1715+func validatePullRequestUpdate(root, update event.Event) error {
1716+ if err := event.Validate(update); err != nil {
1717+ return err
1718+ }
1719+ if update.Kind != event.KIND_GIT_PR_UPDATE {
1720+ return fmt.Errorf("unsupported: event kind %d is not a pull request update", update.Kind)
1721+ }
1722+ parent := event.Tag(update, "E")
1723+ if parent == "" {
1724+ parent = event.Tag(update, "e")
1725+ }
1726+ if parent != root.ID {
1727+ return errors.New("blocked: pull request update does not reference its root")
1728+ }
1729+ declaredAuthor := event.Tag(update, "P")
1730+ if declaredAuthor != root.PubKey {
1731+ return errors.New("blocked: pull request update author tag does not match root")
1732+ }
1733+ if update.PubKey != root.PubKey {
1734+ return errors.New("blocked: pull request update signer is not the PR author")
1735+ }
1736+ if !isObjectID(event.Tag(update, "c")) {
1737+ return errors.New("invalid: pull request update commit is not a SHA-1")
1738+ }
1739+ return nil
1740+}
1741+
1742+func uniqueStrings(values []string) []string {
1743+ seen := make(map[string]struct{}, len(values))
1744+ out := make([]string, 0, len(values))
1745+ for _, value := range values {
1746+ if _, ok := seen[value]; ok {
1747+ continue
1748+ }
1749+ seen[value] = struct{}{}
1750+ out = append(out, value)
1751+ }
1752+ return out
1753+}
1754diff --git a/internal/gitrelay/pr_repair_test.go b/internal/gitrelay/pr_repair_test.go
1755new file mode 100644
1756index 0000000..4e05e04
1757--- /dev/null
1758+++ b/internal/gitrelay/pr_repair_test.go
1759@@ -0,0 +1,249 @@
1760+package gitrelay
1761+
1762+import (
1763+ "context"
1764+ "crypto/x509"
1765+ "encoding/pem"
1766+ "net/http"
1767+ "net/http/httptest"
1768+ "os"
1769+ "os/exec"
1770+ "path/filepath"
1771+ "strings"
1772+ "testing"
1773+
1774+ "github.com/FelineStateMachine/tinyrelay/internal/event"
1775+ "github.com/FelineStateMachine/tinyrelay/internal/policy"
1776+ "github.com/FelineStateMachine/tinyrelay/internal/storage"
1777+)
1778+
1779+func TestPullRequestTargetUsesEventIDForStandardNIP34(t *testing.T) {
1780+ secret := strings.Repeat("0", 63) + "1"
1781+ root := event.Event{Kind: event.KIND_GIT_PR, CreatedAt: 1, Content: "change", Tags: [][]string{{"a", "30617:owner:repo"}, {"c", strings.Repeat("a", 40)}, {"clone", "https://git.example/repo.git"}}}
1782+ if err := event.Sign(&root, secret); err != nil {
1783+ t.Fatal(err)
1784+ }
1785+ r, err := pullRequestTarget(root, event.KIND_GIT_PR)
1786+ if err != nil {
1787+ t.Fatal(err)
1788+ }
1789+ if r.Identifier != root.ID || r.Refs["refs/nostr/"+root.ID] != strings.Repeat("a", 40) {
1790+ t.Fatalf("standard PR target = %#v", r)
1791+ }
1792+}
1793+
1794+func TestPullRequestUpdateRequiresRootAuthor(t *testing.T) {
1795+ rootSecret := strings.Repeat("0", 63) + "1"
1796+ otherSecret := strings.Repeat("0", 63) + "2"
1797+ root := event.Event{Kind: event.KIND_GIT_PR, CreatedAt: 1, Tags: [][]string{{"a", "30617:owner:repo"}, {"c", strings.Repeat("a", 40)}, {"clone", "https://git.example/repo.git"}}}
1798+ if err := event.Sign(&root, rootSecret); err != nil {
1799+ t.Fatal(err)
1800+ }
1801+ update := event.Event{Kind: event.KIND_GIT_PR_UPDATE, CreatedAt: 2, Tags: [][]string{{"E", root.ID}, {"P", root.PubKey}, {"c", strings.Repeat("b", 40)}, {"clone", "https://git.example/repo.git"}}}
1802+ if err := event.Sign(&update, otherSecret); err != nil {
1803+ t.Fatal(err)
1804+ }
1805+ if err := validatePullRequestUpdate(root, update); err == nil {
1806+ t.Fatal("accepted update signed by another author")
1807+ }
1808+}
1809+
1810+func TestPullRequestRepairRejectsMissingCloneBeforeGit(t *testing.T) {
1811+ secret := strings.Repeat("0", 63) + "1"
1812+ root := event.Event{Kind: event.KIND_GIT_PR, CreatedAt: 1, Tags: [][]string{{"a", "30617:owner:repo"}, {"c", strings.Repeat("a", 40)}}}
1813+ owner, err := event.PublicKey(secret)
1814+ if err != nil {
1815+ t.Fatal(err)
1816+ }
1817+ root.Tags[0][1] = "30617:" + owner + ":repo"
1818+ if err := event.Sign(&root, secret); err != nil {
1819+ t.Fatal(err)
1820+ }
1821+ g := &GitRelay{}
1822+ hosted := Repository{Owner: root.PubKey, Identifier: "repo", Refs: map[string]string{}}
1823+ err = g.RepairPullRequestObjects(context.Background(), hosted, PullRequestRepair{Root: root})
1824+ if err == nil || !strings.Contains(err.Error(), "no clone source") {
1825+ t.Fatalf("missing clone error = %v", err)
1826+ }
1827+}
1828+
1829+func TestPullRequestRepairUsesHostedRepositoryRefs(t *testing.T) {
1830+ rootDir := t.TempDir()
1831+ remote := filepath.Join(rootDir, "remote", "repo.git")
1832+ work := filepath.Join(rootDir, "work")
1833+ prGitRun(t, "init", "--bare", remote)
1834+ prGitRun(t, "init", work)
1835+ prGitRun(t, "-C", work, "config", "user.email", "pr@example.test")
1836+ prGitRun(t, "-C", work, "config", "user.name", "PR")
1837+ if err := os.WriteFile(filepath.Join(work, "README"), []byte("one\n"), 0600); err != nil {
1838+ t.Fatal(err)
1839+ }
1840+ prGitRun(t, "-C", work, "add", "README")
1841+ prGitRun(t, "-C", work, "commit", "-m", "one")
1842+ first := strings.TrimSpace(prGitOutput(t, "-C", work, "rev-parse", "HEAD"))
1843+ prGitRun(t, "-C", work, "push", remote, "HEAD:refs/heads/main")
1844+ if err := os.WriteFile(filepath.Join(work, "NEXT"), []byte("two\n"), 0600); err != nil {
1845+ t.Fatal(err)
1846+ }
1847+ prGitRun(t, "-C", work, "add", "NEXT")
1848+ prGitRun(t, "-C", work, "commit", "-m", "two")
1849+ second := strings.TrimSpace(prGitOutput(t, "-C", work, "rev-parse", "HEAD"))
1850+ prGitRun(t, "-C", work, "push", remote, "HEAD:refs/heads/pr")
1851+
1852+ server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { serveGitBackend(t, w, r, filepath.Dir(remote)) }))
1853+ t.Cleanup(server.Close)
1854+ cert, err := x509.ParseCertificate(server.Certificate().Raw)
1855+ if err != nil {
1856+ t.Fatal(err)
1857+ }
1858+ ca := filepath.Join(rootDir, "ca.pem")
1859+ if err := os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Raw}), 0600); err != nil {
1860+ t.Fatal(err)
1861+ }
1862+ t.Setenv("GIT_SSL_CAINFO", ca)
1863+ store, err := storage.Open(context.Background(), filepath.Join(rootDir, "relay.db"))
1864+ if err != nil {
1865+ t.Fatal(err)
1866+ }
1867+ t.Cleanup(func() { _ = store.Close() })
1868+ p := policy.Defaults("")
1869+ g, err := New(Config{Store: store, Root: filepath.Join(rootDir, "git"), Policy: func() policy.Policy { return p }, AllowPrivateRelays: true})
1870+ if err != nil {
1871+ t.Fatal(err)
1872+ }
1873+ secret := strings.Repeat("0", 63) + "1"
1874+ owner, err := event.PublicKey(secret)
1875+ if err != nil {
1876+ t.Fatal(err)
1877+ }
1878+ clone := server.URL + "/repo.git"
1879+ root := event.Event{Kind: event.KIND_GIT_PR, CreatedAt: 1, Content: "PR", Tags: [][]string{{"a", "30617:" + owner + ":repo"}, {"c", first}, {"clone", clone}}}
1880+ if err := event.Sign(&root, secret); err != nil {
1881+ t.Fatal(err)
1882+ }
1883+ update := event.Event{Kind: event.KIND_GIT_PR_UPDATE, CreatedAt: 2, Tags: [][]string{{"E", root.ID}, {"P", owner}, {"c", second}, {"clone", clone}}}
1884+ if err := event.Sign(&update, secret); err != nil {
1885+ t.Fatal(err)
1886+ }
1887+ hosted := Repository{Owner: owner, Identifier: "repo", Clone: []string{clone}, Refs: map[string]string{"refs/heads/main": first}}
1888+ if err := g.FetchMissing(context.Background(), hosted, hosted.Clone, hosted.Refs); err != nil {
1889+ t.Fatal(err)
1890+ }
1891+ if g.objectsPresent(context.Background(), hosted, []string{second}) {
1892+ t.Fatal("PR update was already present before repair; test must exercise a real fetch")
1893+ }
1894+ if err := g.RepairPullRequestObjects(context.Background(), hosted, PullRequestRepair{Root: root, Updates: []event.Event{update}}); err != nil {
1895+ t.Fatal(err)
1896+ }
1897+ gitDir := filepath.Join(rootDir, "git", owner, "repo.git")
1898+ for ref, want := range map[string]string{"refs/heads/main": first, "refs/nostr/" + root.ID: first, "refs/nostr/" + update.ID: second} {
1899+ got := strings.TrimSpace(prGitOutput(t, "--git-dir", gitDir, "rev-parse", ref))
1900+ if got != want {
1901+ t.Fatalf("%s = %s, want %s", ref, got, want)
1902+ }
1903+ }
1904+ if _, err := os.Stat(filepath.Join(gitDir, "tinyrelay.refs")); !os.IsNotExist(err) {
1905+ t.Fatalf("PR repair wrote signed state file: %v", err)
1906+ }
1907+}
1908+
1909+func TestPrivatePolicyDoesNotContactUnconfiguredRepairSource(t *testing.T) {
1910+ var requests int
1911+ server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
1912+ requests++
1913+ w.WriteHeader(http.StatusInternalServerError)
1914+ }))
1915+ t.Cleanup(server.Close)
1916+ root := t.TempDir()
1917+ cert, err := x509.ParseCertificate(server.Certificate().Raw)
1918+ if err != nil {
1919+ t.Fatal(err)
1920+ }
1921+ ca := filepath.Join(root, "ca.pem")
1922+ if err := os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Raw}), 0600); err != nil {
1923+ t.Fatal(err)
1924+ }
1925+ t.Setenv("GIT_SSL_CAINFO", ca)
1926+ store, err := storage.Open(context.Background(), filepath.Join(root, "relay.db"))
1927+ if err != nil {
1928+ t.Fatal(err)
1929+ }
1930+ t.Cleanup(func() { _ = store.Close() })
1931+ p := policy.Defaults("")
1932+ p.Reads = "members"
1933+ g, err := New(Config{Store: store, Root: filepath.Join(root, "git"), Policy: func() policy.Policy { return p }, AllowPrivateRelays: true})
1934+ if err != nil {
1935+ t.Fatal(err)
1936+ }
1937+ repo := Repository{Owner: strings.Repeat("a", 64), Identifier: "private", Clone: []string{server.URL + "/repo.git"}, Refs: map[string]string{"refs/heads/main": strings.Repeat("b", 40)}}
1938+ err = g.FetchMissing(context.Background(), repo, repo.Clone, repo.Refs)
1939+ if err == nil {
1940+ t.Fatal("private-policy repair unexpectedly succeeded")
1941+ }
1942+ if requests != 0 {
1943+ t.Fatalf("private-policy repair contacted unconfigured source %d time(s)", requests)
1944+ }
1945+}
1946+
1947+func TestGRASPTickResumesAfterCanceledRepository(t *testing.T) {
1948+ p := policy.Policy{Features: policy.Features{Grasp: true, Grasp02: true}}
1949+ owner := strings.Repeat("a", 64)
1950+ first := Repository{Owner: owner, Identifier: "a", Refs: map[string]string{}}
1951+ second := Repository{Owner: owner, Identifier: "b", Refs: map[string]string{}}
1952+ var calls []string
1953+ var cancel context.CancelFunc
1954+ g := &GitRelay{
1955+ root: t.TempDir(),
1956+ policy: func() policy.Policy { return p },
1957+ repos: map[string]Repository{key(first.Owner, first.Identifier): first, key(second.Owner, second.Identifier): second},
1958+ gitSync: func(ctx context.Context, repo Repository) error {
1959+ calls = append(calls, repo.Identifier)
1960+ if len(calls) == 1 {
1961+ cancel()
1962+ return ctx.Err()
1963+ }
1964+ return nil
1965+ },
1966+ }
1967+ ctx, done := context.WithCancel(context.Background())
1968+ cancel = done
1969+ if err := g.GRASPService().Tick(ctx); err == nil {
1970+ t.Fatal("canceled GRASP tick unexpectedly succeeded")
1971+ }
1972+ status, err := g.GRASPService().SyncStatus(context.Background())
1973+ if err != nil {
1974+ t.Fatal(err)
1975+ }
1976+ if _, ok := status.Repositories[key(second.Owner, second.Identifier)]; ok {
1977+ t.Fatalf("unattempted repository received a status: %#v", status.Repositories)
1978+ }
1979+ if err := g.GRASPService().Tick(context.Background()); err != nil {
1980+ t.Fatal(err)
1981+ }
1982+ status, err = g.GRASPService().SyncStatus(context.Background())
1983+ if err != nil {
1984+ t.Fatal(err)
1985+ }
1986+ if status.Repositories[key(second.Owner, second.Identifier)].SucceededAt == 0 {
1987+ t.Fatalf("resumed repository was not attempted: %#v", status.Repositories)
1988+ }
1989+ if len(calls) != 2 || calls[1] != "b" {
1990+ t.Fatalf("resume order = %v, want [a b]", calls)
1991+ }
1992+}
1993+
1994+func prGitRun(t *testing.T, args ...string) {
1995+ t.Helper()
1996+ if out, err := exec.Command("git", args...).CombinedOutput(); err != nil {
1997+ t.Fatalf("git %v: %v: %s", args, err, out)
1998+ }
1999+}
2000+
2001+func prGitOutput(t *testing.T, args ...string) string {
2002+ t.Helper()
2003+ out, err := exec.Command("git", args...).Output()
2004+ if err != nil {
2005+ t.Fatal(err)
2006+ }
2007+ return string(out)
2008+}
2009diff --git a/internal/replication/sync_acceptance_test.go b/internal/replication/sync_acceptance_test.go
2010index 74863fd..9a29790 100644
2011--- a/internal/replication/sync_acceptance_test.go
2012+++ b/internal/replication/sync_acceptance_test.go
2013@@ -182,6 +182,57 @@ func TestQueryNegentropyEqualInventoryCloseCompletesImmediately(t *testing.T) {
2014 }
2015 }
2016
2017+type negRejectDialer struct{ calls int }
2018+
2019+type negRejectSocket struct{ responses [][]byte }
2020+
2021+func (d *negRejectDialer) Dial(context.Context, string) (Socket, *http.Response, error) {
2022+ d.calls++
2023+ return &negRejectSocket{}, nil, nil
2024+}
2025+
2026+func (s *negRejectSocket) Read(ctx context.Context) ([]byte, error) {
2027+ if len(s.responses) == 0 {
2028+ return nil, ctx.Err()
2029+ }
2030+ item := s.responses[0]
2031+ s.responses = s.responses[1:]
2032+ return item, nil
2033+}
2034+
2035+func (s *negRejectSocket) Write(_ context.Context, data []byte) error {
2036+ var message []json.RawMessage
2037+ if err := json.Unmarshal(data, &message); err != nil || len(message) < 2 {
2038+ return nil
2039+ }
2040+ var kind, id string
2041+ _ = json.Unmarshal(message[0], &kind)
2042+ _ = json.Unmarshal(message[1], &id)
2043+ var response []any
2044+ if kind == "NEG-OPEN" {
2045+ response = []any{"NEG-ERR", id, "unsupported"}
2046+ } else if kind == "REQ" {
2047+ response = []any{"EOSE", id}
2048+ }
2049+ raw, _ := json.Marshal(response)
2050+ s.responses = append(s.responses, raw)
2051+ return nil
2052+}
2053+
2054+func (s *negRejectSocket) Close(error) error { return nil }
2055+
2056+func TestQuerySynchronizedFallsBackWhenNegentropyIsUnsupported(t *testing.T) {
2057+ dialer := &negRejectDialer{}
2058+ transport := &NostrTransport{Dialer: dialer, Timeout: time.Second}
2059+ items, err := transport.QuerySynchronized(context.Background(), "ws://relay.example", event.Filter{}, nil)
2060+ if err != nil {
2061+ t.Fatal(err)
2062+ }
2063+ if len(items) != 0 || dialer.calls != 2 {
2064+ t.Fatalf("fallback result=%d dials=%d, want empty result and two dials", len(items), dialer.calls)
2065+ }
2066+}
2067+
2068 func TestNegentropyNeededIDsAreFetchedInBoundedFilteredBatches(t *testing.T) {
2069 const total = negentropyFetchBatch + 1
2070 secret := "1111111111111111111111111111111111111111111111111111111111111111"
2071diff --git a/internal/replication/websocket.go b/internal/replication/websocket.go
2072index da303aa..d11d067 100644
2073--- a/internal/replication/websocket.go
2074+++ b/internal/replication/websocket.go
2075@@ -8,6 +8,8 @@ import (
2076 "net"
2077 "net/http"
2078 "net/url"
2079+ "strings"
2080+ "sync"
2081 "sync/atomic"
2082 "time"
2083
2084@@ -252,6 +254,221 @@ type NostrTransport struct {
2085 Dialer Dialer
2086 Timeout time.Duration
2087 sequence atomic.Uint64
2088+ legacyMu sync.Mutex
2089+ legacy map[string]struct{}
2090+}
2091+
2092+// Subscribe keeps a bounded NIP-01 subscription open until ctx is canceled.
2093+// Reconnection belongs to the lifecycle owner so it can apply peer backoff
2094+// and avoid multiplying sockets when a relay is unhealthy.
2095+func (t *NostrTransport) Subscribe(ctx context.Context, target string, filter event.Filter, onEvent func(event.Event) error) error {
2096+ return t.SubscribeFilters(ctx, target, []event.Filter{filter}, onEvent)
2097+}
2098+
2099+// SubscribeFilters opens one subscription for a set of NIP-01 filters. Relays
2100+// treat filters in one REQ as an OR, which lets callers follow related event
2101+// families without opening a socket for each family.
2102+func (t *NostrTransport) SubscribeFilters(ctx context.Context, target string, filters []event.Filter, onEvent func(event.Event) error) error {
2103+ if len(filters) == 0 {
2104+ return errors.New("replication: live subscription has no filters")
2105+ }
2106+ socket, _, err := t.dial(ctx, target)
2107+ if err != nil {
2108+ return err
2109+ }
2110+ defer socket.Close(nil)
2111+ sub := fmt.Sprintf("tiny-live-%d", t.sequence.Add(1))
2112+ prepared := make([]event.Filter, len(filters))
2113+ for i, filter := range filters {
2114+ filter.Tags = ensureTags(filter.Tags)
2115+ prepared[i] = filter
2116+ }
2117+ message, err := json.Marshal(append([]any{"REQ", sub}, filtersToAny(prepared)...))
2118+ if err != nil {
2119+ return fmt.Errorf("encode subscription: %w", err)
2120+ }
2121+ if err := socket.Write(ctx, message); err != nil {
2122+ return err
2123+ }
2124+ for {
2125+ data, err := socket.Read(ctx)
2126+ if err != nil {
2127+ return err
2128+ }
2129+ var raw []json.RawMessage
2130+ if json.Unmarshal(data, &raw) != nil || len(raw) < 2 {
2131+ continue
2132+ }
2133+ var kind, id string
2134+ _ = json.Unmarshal(raw[0], &kind)
2135+ _ = json.Unmarshal(raw[1], &id)
2136+ if id != sub {
2137+ continue
2138+ }
2139+ if kind == "CLOSED" {
2140+ return errors.New("replication: live subscription closed")
2141+ }
2142+ if kind != "EVENT" || len(raw) < 3 {
2143+ continue
2144+ }
2145+ var item event.Event
2146+ if err := json.Unmarshal(raw[2], &item); err != nil {
2147+ continue
2148+ }
2149+ if err := event.Validate(item); err != nil || !filterMatchesAny(prepared, item) {
2150+ continue
2151+ }
2152+ if onEvent != nil {
2153+ if err := onEvent(item); err != nil {
2154+ return err
2155+ }
2156+ }
2157+ }
2158+}
2159+
2160+func filtersToAny(filters []event.Filter) []any {
2161+ out := make([]any, len(filters))
2162+ for i := range filters {
2163+ out[i] = filters[i]
2164+ }
2165+ return out
2166+}
2167+
2168+func filterMatchesAny(filters []event.Filter, item event.Event) bool {
2169+ for _, filter := range filters {
2170+ if event.Matches(filter, item) {
2171+ return true
2172+ }
2173+ }
2174+ return false
2175+}
2176+
2177+const (
2178+ // pullPageSize keeps a compatibility REQ small enough for relays to serve
2179+ // without allocating an unbounded result set.
2180+ pullPageSize = 500
2181+ // pullPageLimit prevents a corrupt or very large repository announcement
2182+ // from turning one maintenance pass into an unbounded import.
2183+ pullPageLimit = 20
2184+ pullEventLimit = pullPageSize * pullPageLimit
2185+)
2186+
2187+// ErrPullIncomplete indicates that a relay returned a full bounded history
2188+// window but the next page cannot be represented safely by an `until` cursor.
2189+// Callers should retain the result and retry with a larger or more capable
2190+// synchronization method.
2191+var ErrPullIncomplete = errors.New("replication: pull history is incomplete")
2192+
2193+// ErrNegentropyUnsupported means the peer does not implement NIP-77. It is
2194+// deliberately distinct from authentication, protocol and transport errors:
2195+// only this condition is safe to downgrade to an ordinary REQ pull.
2196+var ErrNegentropyUnsupported = errors.New("replication: negentropy unsupported")
2197+
2198+// QuerySynchronized prefers NIP-77 negentropy and falls back to bounded,
2199+// descending REQ pages when the peer does not implement it. GRASP peers are
2200+// commonly mixed-version, so lack of NIP-77 support must not prevent the
2201+// ordinary NIP-01 pull from completing.
2202+func (t *NostrTransport) QuerySynchronized(ctx context.Context, target string, filter event.Filter, local []syncprotocol.Item) ([]event.Event, error) {
2203+ if t.knownLegacy(target) {
2204+ return t.QueryPaginated(ctx, target, filter)
2205+ }
2206+ items, err := t.QueryNegentropy(ctx, target, filter, local)
2207+ if err == nil {
2208+ return items, nil
2209+ }
2210+ if !errors.Is(err, ErrNegentropyUnsupported) {
2211+ return items, err
2212+ }
2213+ t.rememberLegacy(target)
2214+ return t.QueryPaginated(ctx, target, filter)
2215+}
2216+
2217+func (t *NostrTransport) knownLegacy(target string) bool {
2218+ t.legacyMu.Lock()
2219+ defer t.legacyMu.Unlock()
2220+ _, ok := t.legacy[target]
2221+ return ok
2222+}
2223+
2224+func (t *NostrTransport) rememberLegacy(target string) {
2225+ t.legacyMu.Lock()
2226+ defer t.legacyMu.Unlock()
2227+ if t.legacy == nil {
2228+ t.legacy = make(map[string]struct{})
2229+ }
2230+ // Bound the instance-local compatibility cache so untrusted relay URLs
2231+ // cannot accumulate indefinitely. Once full, an unremembered peer simply
2232+ // receives another bounded probe on a later sync.
2233+ if len(t.legacy) < 64 {
2234+ t.legacy[target] = struct{}{}
2235+ }
2236+}
2237+
2238+// QueryPaginated reads a bounded history window. It uses the oldest event in
2239+// each page as the next upper bound, deduplicating events because relays may
2240+// include the boundary event in their response.
2241+func (t *NostrTransport) QueryPaginated(ctx context.Context, target string, filter event.Filter) ([]event.Event, error) {
2242+ result := make([]event.Event, 0, pullPageSize)
2243+ seen := make(map[string]struct{})
2244+ pageFilter := filter
2245+ for page := 0; page < pullPageLimit; page++ {
2246+ if err := ctx.Err(); err != nil {
2247+ return result, err
2248+ }
2249+ limit := pullPageSize
2250+ pageFilter.Limit = &limit
2251+ items, err := t.Query(ctx, target, pageFilter)
2252+ added := 0
2253+ for _, item := range items {
2254+ if _, ok := seen[item.ID]; ok {
2255+ continue
2256+ }
2257+ seen[item.ID] = struct{}{}
2258+ result = append(result, item)
2259+ added++
2260+ }
2261+ if len(result) >= pullEventLimit {
2262+ return result, ErrPullIncomplete
2263+ }
2264+ if err != nil {
2265+ return result, err
2266+ }
2267+ if len(items) == 0 {
2268+ return result, nil
2269+ }
2270+ oldest := items[0].CreatedAt
2271+ for _, item := range items {
2272+ if item.CreatedAt < oldest {
2273+ oldest = item.CreatedAt
2274+ }
2275+ }
2276+ // The inclusive boundary is an overlap probe. If it yields no new
2277+ // events, the relay has no older matching events available through this
2278+ // cursor; a short response is therefore not treated as definitive by
2279+ // itself.
2280+ if page > 0 && added == 0 {
2281+ allAtBoundary := true
2282+ for _, item := range items {
2283+ if item.CreatedAt != oldest {
2284+ allAtBoundary = false
2285+ break
2286+ }
2287+ }
2288+ if allAtBoundary && len(items) >= pullPageSize {
2289+ return result, ErrPullIncomplete
2290+ }
2291+ return result, nil
2292+ }
2293+ if oldest <= 0 {
2294+ return result, nil
2295+ }
2296+ // Keep the boundary inclusive. A relay may cap every response below
2297+ // our requested limit, and a strict cursor would silently drop events
2298+ // sharing the oldest timestamp. Deduplication makes the overlap safe.
2299+ until := oldest
2300+ pageFilter.Until = &until
2301+ }
2302+ return result, ErrPullIncomplete
2303 }
2304
2305 // QueryNegentropy reconciles a local index with a remote relay, then fetches
2306@@ -281,11 +498,19 @@ func (t *NostrTransport) QueryNegentropy(ctx context.Context, target string, fil
2307 return nil, err
2308 }
2309 var needed []string
2310+ progress := false
2311 deadline, cancel := context.WithTimeout(ctx, t.timeout())
2312 defer cancel()
2313 for {
2314 data, err := socket.Read(deadline)
2315 if err != nil {
2316+ // Some legacy relays accept the websocket but silently ignore
2317+ // NEG-OPEN. A bounded probe timeout is the only safe downgrade; a
2318+ // caller cancellation or an authenticated/protocol response must
2319+ // remain an error.
2320+ if errors.Is(err, context.DeadlineExceeded) && !progress && ctx.Err() == nil {
2321+ return nil, ErrNegentropyUnsupported
2322+ }
2323 return nil, err
2324 }
2325 var raw []json.RawMessage
2326@@ -298,6 +523,9 @@ func (t *NostrTransport) QueryNegentropy(ctx context.Context, target string, fil
2327 if len(raw) > 2 {
2328 _ = json.Unmarshal(raw[2], &body)
2329 }
2330+ if kind == "AUTH" {
2331+ return nil, errors.New("replication: remote relay requires authentication")
2332+ }
2333 if messageID != id {
2334 continue
2335 }
2336@@ -306,19 +534,30 @@ func (t *NostrTransport) QueryNegentropy(ctx context.Context, target string, fil
2337 // waiting for a NEG-MSG that will never arrive turns a successful NIP-77
2338 // no-op sync into a timeout.
2339 if kind == "NEG-CLOSE" {
2340+ progress = true
2341 break
2342 }
2343+ if kind == "CLOSED" {
2344+ return nil, errors.New("replication: remote negentropy closed")
2345+ }
2346 if kind == "NEG-ERR" {
2347- return nil, errors.New("replication: remote negentropy rejected")
2348+ if negentropyUnsupported(body) {
2349+ return nil, ErrNegentropyUnsupported
2350+ }
2351+ return nil, fmt.Errorf("replication: remote negentropy rejected: %s", body)
2352 }
2353 if kind != "NEG-MSG" {
2354 continue
2355 }
2356+ progress = true
2357 result, err := session.Reconcile(body)
2358 if err != nil {
2359 return nil, err
2360 }
2361 needed = append(needed, result.Need...)
2362+ if len(needed) > pullEventLimit {
2363+ return nil, ErrPullIncomplete
2364+ }
2365 if result.Done {
2366 closeMessage, _ := json.Marshal([]any{"NEG-CLOSE", id})
2367 _ = socket.Write(ctx, closeMessage)
2368@@ -338,6 +577,19 @@ func (t *NostrTransport) QueryNegentropy(ctx context.Context, target string, fil
2369 return t.queryNegentropyIDs(ctx, target, filter, needed)
2370 }
2371
2372+func negentropyUnsupported(reason string) bool {
2373+ reason = strings.ToLower(strings.TrimSpace(reason))
2374+ if strings.HasPrefix(reason, "unsupported") || strings.HasPrefix(reason, "error: unsupported") {
2375+ return true
2376+ }
2377+ switch reason {
2378+ case "unsupported", "not supported", "negentropy unsupported", "nip-77 unsupported":
2379+ return true
2380+ default:
2381+ return false
2382+ }
2383+}
2384+
2385 const negentropyFetchBatch = 500
2386
2387 func (t *NostrTransport) queryNegentropyIDs(ctx context.Context, target string, filter event.Filter, ids []string) ([]event.Event, error) {
2388@@ -355,7 +607,14 @@ func (t *NostrTransport) queryNegentropyIDs(ctx context.Context, target string,
2389 request.Limit = &limit
2390 events, err := t.Query(ctx, target, request)
2391 if err != nil {
2392- return nil, err
2393+ for _, candidate := range events {
2394+ if _, ok := seen[candidate.ID]; ok {
2395+ continue
2396+ }
2397+ seen[candidate.ID] = struct{}{}
2398+ result = append(result, candidate)
2399+ }
2400+ return result, err
2401 }
2402 allowed := make(map[string]struct{}, len(batch))
2403 for _, id := range batch {
2404@@ -375,9 +634,20 @@ func (t *NostrTransport) queryNegentropyIDs(ctx context.Context, target string,
2405 result = append(result, candidate)
2406 }
2407 }
2408+ if len(seen) != len(uniqueIDs(ids)) {
2409+ return result, ErrPullIncomplete
2410+ }
2411 return result, nil
2412 }
2413
2414+func uniqueIDs(ids []string) map[string]struct{} {
2415+ unique := make(map[string]struct{}, len(ids))
2416+ for _, id := range ids {
2417+ unique[id] = struct{}{}
2418+ }
2419+ return unique
2420+}
2421+
2422 func (t *NostrTransport) timeout() time.Duration {
2423 if t.Timeout > 0 {
2424 return t.Timeout
2425@@ -400,7 +670,7 @@ func (t *NostrTransport) Query(ctx context.Context, target string, filter event.
2426 if err := socket.Write(ctx, message); err != nil {
2427 return nil, err
2428 }
2429- return readEvents(ctx, socket, sub, t.timeout())
2430+ return readEvents(ctx, socket, sub, t.timeout(), filter)
2431 }
2432
2433 func (t *NostrTransport) Send(ctx context.Context, target string, e event.Event) (DeliveryResult, error) {
2434@@ -447,14 +717,14 @@ func (t *NostrTransport) dial(ctx context.Context, target string) (Socket, *http
2435 return dialer.Dial(ctx, target)
2436 }
2437
2438-func readEvents(ctx context.Context, socket Socket, subscription string, timeout time.Duration) ([]event.Event, error) {
2439+func readEvents(ctx context.Context, socket Socket, subscription string, timeout time.Duration, filter event.Filter) ([]event.Event, error) {
2440 deadline, cancel := context.WithTimeout(ctx, timeout)
2441 defer cancel()
2442 items := make([]event.Event, 0)
2443 for {
2444 data, err := socket.Read(deadline)
2445 if err != nil {
2446- return nil, err
2447+ return items, err
2448 }
2449 var raw []json.RawMessage
2450 if err := json.Unmarshal(data, &raw); err != nil || len(raw) < 2 {
2451@@ -469,8 +739,11 @@ func readEvents(ctx context.Context, socket Socket, subscription string, timeout
2452 continue
2453 }
2454 var item event.Event
2455- if json.Unmarshal(raw[2], &item) == nil && event.Validate(item) == nil {
2456+ if json.Unmarshal(raw[2], &item) == nil && event.Validate(item) == nil && event.Matches(filter, item) {
2457 items = append(items, item)
2458+ if len(items) > pullEventLimit {
2459+ return items[:pullEventLimit], ErrPullIncomplete
2460+ }
2461 }
2462 case "EOSE":
2463 if id == subscription {
2464@@ -478,10 +751,10 @@ func readEvents(ctx context.Context, socket Socket, subscription string, timeout
2465 }
2466 case "CLOSED":
2467 if id == subscription {
2468- return nil, errors.New("replication: remote query closed")
2469+ return items, errors.New("replication: remote query closed")
2470 }
2471 case "AUTH":
2472- return nil, errors.New("replication: remote relay requires authentication")
2473+ return items, errors.New("replication: remote relay requires authentication")
2474 }
2475 }
2476 }
2477diff --git a/internal/replication/websocket_transport_test.go b/internal/replication/websocket_transport_test.go
2478new file mode 100644
2479index 0000000..631e4d1
2480--- /dev/null
2481+++ b/internal/replication/websocket_transport_test.go
2482@@ -0,0 +1,87 @@
2483+package replication
2484+
2485+import (
2486+ "context"
2487+ "encoding/json"
2488+ "net/http"
2489+ "testing"
2490+ "time"
2491+
2492+ "github.com/FelineStateMachine/tinyrelay/internal/event"
2493+)
2494+
2495+type silentNegentropyDialer struct{ calls int }
2496+
2497+type silentNegentropySocket struct{ responses chan []byte }
2498+
2499+func (d *silentNegentropyDialer) Dial(_ context.Context, _ string) (Socket, *http.Response, error) {
2500+ d.calls++
2501+ return &silentNegentropySocket{responses: make(chan []byte, 1)}, nil, nil
2502+}
2503+
2504+func (s *silentNegentropySocket) Read(ctx context.Context) ([]byte, error) {
2505+ select {
2506+ case response := <-s.responses:
2507+ return response, nil
2508+ case <-ctx.Done():
2509+ return nil, ctx.Err()
2510+ }
2511+}
2512+
2513+func (s *silentNegentropySocket) Write(_ context.Context, data []byte) error {
2514+ var message []json.RawMessage
2515+ if err := json.Unmarshal(data, &message); err != nil || len(message) < 2 {
2516+ return nil
2517+ }
2518+ var kind, id string
2519+ _ = json.Unmarshal(message[0], &kind)
2520+ _ = json.Unmarshal(message[1], &id)
2521+ if kind != "REQ" {
2522+ return nil // legacy peer silently ignores NEG-OPEN
2523+ }
2524+ response, _ := json.Marshal([]any{"EOSE", id})
2525+ s.responses <- response
2526+ return nil
2527+}
2528+
2529+func (s *silentNegentropySocket) Close(error) error { return nil }
2530+
2531+func TestQuerySynchronizedFallsBackAfterSilentLegacyNegentropyPeer(t *testing.T) {
2532+ dialer := &silentNegentropyDialer{}
2533+ transport := &NostrTransport{Dialer: dialer, Timeout: 10 * time.Millisecond}
2534+ items, err := transport.QuerySynchronized(context.Background(), "ws://relay.example", event.Filter{}, nil)
2535+ if err != nil {
2536+ t.Fatal(err)
2537+ }
2538+ if len(items) != 0 {
2539+ t.Fatalf("fallback returned %d events", len(items))
2540+ }
2541+ if dialer.calls != 2 {
2542+ t.Fatalf("transport dials = %d, want negentropy probe plus REQ fallback", dialer.calls)
2543+ }
2544+}
2545+
2546+func TestQuerySynchronizedCachesSilentLegacyPeerPerTransport(t *testing.T) {
2547+ dialer := &silentNegentropyDialer{}
2548+ transport := &NostrTransport{Dialer: dialer, Timeout: 10 * time.Millisecond}
2549+ for i := 0; i < 2; i++ {
2550+ if _, err := transport.QuerySynchronized(context.Background(), "ws://legacy.example", event.Filter{}, nil); err != nil {
2551+ t.Fatal(err)
2552+ }
2553+ }
2554+ if dialer.calls != 3 {
2555+ t.Fatalf("transport dials = %d, want one negentropy probe and two REQ pulls", dialer.calls)
2556+ }
2557+}
2558+
2559+func TestQueryNegentropyIDsReportsMissingRequestedEvents(t *testing.T) {
2560+ dialer := &batchQueryDialer{events: map[string]event.Event{}}
2561+ transport := &NostrTransport{Dialer: dialer, Timeout: time.Second}
2562+ items, err := transport.queryNegentropyIDs(context.Background(), "ws://relay.example", event.Filter{}, []string{"missing"})
2563+ if err != ErrPullIncomplete {
2564+ t.Fatalf("error = %v, want ErrPullIncomplete", err)
2565+ }
2566+ if len(items) != 0 {
2567+ t.Fatalf("returned %d events for missing ID", len(items))
2568+ }
2569+}

About

ownerc01503e8035f
visibilitypublic
headf9b06b9bec64

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs