| 1 | diff --git a/internal/daemon/git_services.go b/internal/daemon/git_services.go |
| 2 | index 51e0a40..48a9cd1 100644 |
| 3 | --- a/internal/daemon/git_services.go |
| 4 | +++ b/internal/daemon/git_services.go |
| 5 | @@ -2,6 +2,10 @@ package daemon |
| 6 | |
| 7 | import ( |
| 8 | "context" |
| 9 | + "crypto/sha256" |
| 10 | + "database/sql" |
| 11 | + "encoding/hex" |
| 12 | + "encoding/json" |
| 13 | "errors" |
| 14 | "net/url" |
| 15 | "sort" |
| 16 | @@ -51,6 +55,9 @@ func (t *Tenant) gitEventSync(ctx context.Context, repo gitrelay.Repository) err |
| 17 | result = errors.Join(result, t.gitSyncPeer(peerCtx, repo, target)) |
| 18 | cancel() |
| 19 | } |
| 20 | + if historyCtx.Err() != nil && ctx.Err() == nil { |
| 21 | + result = errors.Join(result, replication.ErrPullIncomplete) |
| 22 | + } |
| 23 | historyCancel() |
| 24 | repairCtx, repairCancel := context.WithTimeout(ctx, 5*time.Second) |
| 25 | result = errors.Join(result, t.repairGitPullRequests(repairCtx, repo)) |
| 26 | @@ -60,13 +67,25 @@ func (t *Tenant) gitEventSync(ctx context.Context, repo gitrelay.Repository) err |
| 27 | result = errors.Join(result, t.gitOutboxSync(outboxCtx, repo)) |
| 28 | outboxCancel() |
| 29 | } |
| 30 | - return withoutBudgetDeadline(ctx, result) |
| 31 | + return gitSyncOutcome(ctx, result) |
| 32 | +} |
| 33 | + |
| 34 | +// gitSyncOutcome classifies a pass for the scheduler. Deadlines from the |
| 35 | +// pass's own time budgets are dropped because peer and outbox offsets persist |
| 36 | +// between passes. A pass whose only remaining condition is more history to |
| 37 | +// fetch reports gitrelay.ErrIncomplete so it continues soon without counting |
| 38 | +// as a failure. Cancellation of the caller's context still propagates. |
| 39 | +func gitSyncOutcome(parent context.Context, err error) error { |
| 40 | + err = withoutBudgetDeadline(parent, err) |
| 41 | + if err == nil { |
| 42 | + return nil |
| 43 | + } |
| 44 | + if onlyPullIncomplete(err) { |
| 45 | + return gitrelay.ErrIncomplete |
| 46 | + } |
| 47 | + return err |
| 48 | } |
| 49 | |
| 50 | -// withoutBudgetDeadline drops deadlines from the pass's own time budgets. |
| 51 | -// Peer offsets and outbox offsets persist between passes, so running out of |
| 52 | -// time leaves the repository incomplete rather than failed and keeps it on the |
| 53 | -// regular cadence. Cancellation of the caller's context still propagates. |
| 54 | func withoutBudgetDeadline(parent context.Context, err error) error { |
| 55 | if err == nil || parent.Err() != nil { |
| 56 | return err |
| 57 | @@ -86,6 +105,31 @@ func withoutBudgetDeadline(parent context.Context, err error) error { |
| 58 | return err |
| 59 | } |
| 60 | |
| 61 | +func onlyPullIncomplete(err error) bool { |
| 62 | + if joined, ok := err.(interface{ Unwrap() []error }); ok { |
| 63 | + for _, item := range joined.Unwrap() { |
| 64 | + if !onlyPullIncomplete(item) { |
| 65 | + return false |
| 66 | + } |
| 67 | + } |
| 68 | + return true |
| 69 | + } |
| 70 | + return errors.Is(err, replication.ErrPullIncomplete) |
| 71 | +} |
| 72 | + |
| 73 | +// gitIngestRejected reports a deterministic admission decision. NIP-01 |
| 74 | +// prefixes mark policy rejections that repeat on every attempt; other errors |
| 75 | +// are storage or lookup failures that may succeed later. |
| 76 | +func gitIngestRejected(err error) bool { |
| 77 | + message := err.Error() |
| 78 | + for _, prefix := range []string{"blocked:", "invalid:", "restricted:", "auth-required:", "pow:"} { |
| 79 | + if strings.HasPrefix(message, prefix) { |
| 80 | + return true |
| 81 | + } |
| 82 | + } |
| 83 | + return false |
| 84 | +} |
| 85 | + |
| 86 | // gitTargetTransport checks the privacy boundary before sending any filters. |
| 87 | // An unverified private peer never falls back to an anonymous connection. |
| 88 | func (t *Tenant) gitTargetTransport(ctx context.Context, repo gitrelay.Repository, target string) (*replication.NostrTransport, error) { |
| 89 | @@ -101,7 +145,7 @@ func (t *Tenant) gitTargetTransport(ctx context.Context, repo gitrelay.Repositor |
| 90 | return nil, errors.New("private peer is not ready") |
| 91 | } |
| 92 | } |
| 93 | - return &replication.NostrTransport{Dialer: t.gitEventDialer(repo.Private || t.PrivateServiceEnabled()), Timeout: 5 * time.Second}, nil |
| 94 | + return &replication.NostrTransport{Dialer: t.gitEventDialer(repo.Private || t.PrivateServiceEnabled()), Timeout: 5 * time.Second, LegacyCache: t.gitLegacy}, nil |
| 95 | } |
| 96 | |
| 97 | func (t *Tenant) gitSyncPeer(ctx context.Context, repo gitrelay.Repository, target string) error { |
| 98 | @@ -125,38 +169,151 @@ func (t *Tenant) gitSyncPeer(ctx context.Context, repo gitrelay.Repository, targ |
| 99 | |
| 100 | func (t *Tenant) gitPullFilter(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) error { |
| 101 | local, err := t.localSyncItems(ctx, filter) |
| 102 | + if errors.Is(err, replication.ErrPullIncomplete) { |
| 103 | + return t.gitPullLargeFilter(ctx, transport, target, filter) |
| 104 | + } |
| 105 | if err != nil { |
| 106 | return err |
| 107 | } |
| 108 | items, pullErr := transport.QuerySynchronized(ctx, target, filter, local) |
| 109 | + return errors.Join(pullErr, t.ingestGitHistory(ctx, filter, items)) |
| 110 | +} |
| 111 | + |
| 112 | +// A partial inventory is not safe for negentropy: the resulting missing-ID set |
| 113 | +// can overflow its fetch cap. Refresh a small head first, then resume an older |
| 114 | +// window. A slow historical query cannot prevent new events from being stored. |
| 115 | +func (t *Tenant) gitPullLargeFilter(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) error { |
| 116 | + olderFilter, hasCursor, err := t.gitHistoryWindow(ctx, target, filter) |
| 117 | + if err != nil { |
| 118 | + return err |
| 119 | + } |
| 120 | + headCtx, cancelHead := gitQueryContext(ctx, 2*time.Second) |
| 121 | + head, headErr := queryGitHead(headCtx, transport, target, filter) |
| 122 | + cancelHead() |
| 123 | + headIngestErr := t.ingestGitHistory(ctx, filter, head) |
| 124 | + result := errors.Join(replication.ErrPullIncomplete, headErr, headIngestErr) |
| 125 | + key := gitHistoryCursorKey(target, filter) |
| 126 | + if !hasCursor { |
| 127 | + // A completed bounded head query seeds history; it never proves that |
| 128 | + // the remote inventory is exhausted. Failed queries cannot move a |
| 129 | + // timestamp cursor because events may have arrived out of order. |
| 130 | + if headErr == nil && headIngestErr == nil && len(head) > 0 { |
| 131 | + result = errors.Join(result, t.store.PutSetting(ctx, key, oldestGitHistoryCursor(head))) |
| 132 | + } |
| 133 | + return result |
| 134 | + } |
| 135 | + if ctx.Err() != nil { |
| 136 | + return errors.Join(result, ctx.Err()) |
| 137 | + } |
| 138 | + olderCtx, cancelOlder := gitQueryContext(ctx, 5*time.Second) |
| 139 | + older, olderErr := transport.QueryPaginated(olderCtx, target, olderFilter) |
| 140 | + cancelOlder() |
| 141 | + olderIngestErr := t.ingestGitHistory(ctx, filter, older) |
| 142 | + result = errors.Join(result, olderErr, olderIngestErr) |
| 143 | + if olderIngestErr != nil { |
| 144 | + return result |
| 145 | + } |
| 146 | + if olderErr == nil { |
| 147 | + // Finished this sweep. The next pass starts another from the head, |
| 148 | + // allowing older events newly acquired by the peer to be discovered. |
| 149 | + result = errors.Join(result, t.store.PutSetting(ctx, key, storage.EventCursor{})) |
| 150 | + } else if errors.Is(olderErr, replication.ErrPullIncomplete) && len(older) > 0 { |
| 151 | + result = errors.Join(result, t.store.PutSetting(ctx, key, oldestGitHistoryCursor(older))) |
| 152 | + } |
| 153 | + return result |
| 154 | +} |
| 155 | + |
| 156 | +func (t *Tenant) ingestGitHistory(ctx context.Context, filter event.Filter, items []event.Event) error { |
| 157 | sort.SliceStable(items, func(i, j int) bool { return metadataEventOrder(items[i].Kind) < metadataEventOrder(items[j].Kind) }) |
| 158 | + var result error |
| 159 | for _, item := range items { |
| 160 | if ctx.Err() != nil { |
| 161 | - return errors.Join(pullErr, ctx.Err()) |
| 162 | + return errors.Join(result, ctx.Err()) |
| 163 | } |
| 164 | if !event.Matches(filter, item) || event.Validate(item) != nil { |
| 165 | continue |
| 166 | } |
| 167 | - // Rejected remote events must not prevent later valid events from arriving. |
| 168 | + // A rejected event cannot prevent later valid events from arriving, |
| 169 | + // and a policy rejection repeats on every attempt, so it counts as |
| 170 | + // processed. A storage failure keeps the checkpoint so the window is |
| 171 | + // retried without skipping an event that was not persisted. |
| 172 | if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) { |
| 173 | t.app.telemetry.Logger().Debug("Git sync event rejected", "kind", item.Kind, "error", err) |
| 174 | + if !gitIngestRejected(err) { |
| 175 | + result = errors.Join(result, err) |
| 176 | + } |
| 177 | + } |
| 178 | + } |
| 179 | + return result |
| 180 | +} |
| 181 | + |
| 182 | +const gitHeadPageSize = 256 |
| 183 | + |
| 184 | +// Reserve time to ingest the response using the parent context. A query that |
| 185 | +// expires may still have returned valid events worth retaining for the retry. |
| 186 | +func gitQueryContext(ctx context.Context, maximum time.Duration) (context.Context, context.CancelFunc) { |
| 187 | + if deadline, ok := ctx.Deadline(); ok { |
| 188 | + if remaining := time.Until(deadline) / 2; remaining < maximum { |
| 189 | + maximum = remaining |
| 190 | } |
| 191 | } |
| 192 | - return pullErr |
| 193 | + return context.WithTimeout(ctx, maximum) |
| 194 | +} |
| 195 | + |
| 196 | +func queryGitHead(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) ([]event.Event, error) { |
| 197 | + limit := gitHeadPageSize |
| 198 | + filter.Limit = &limit |
| 199 | + return transport.Query(ctx, target, filter) |
| 200 | +} |
| 201 | + |
| 202 | +func oldestGitHistoryCursor(items []event.Event) storage.EventCursor { |
| 203 | + if len(items) == 0 { |
| 204 | + return storage.EventCursor{} |
| 205 | + } |
| 206 | + oldest := items[0] |
| 207 | + for _, item := range items[1:] { |
| 208 | + if item.CreatedAt < oldest.CreatedAt || (item.CreatedAt == oldest.CreatedAt && item.ID > oldest.ID) { |
| 209 | + oldest = item |
| 210 | + } |
| 211 | + } |
| 212 | + return storage.EventCursor{CreatedAt: oldest.CreatedAt, ID: oldest.ID} |
| 213 | +} |
| 214 | + |
| 215 | +func gitHistoryCursorKey(target string, filter event.Filter) string { |
| 216 | + raw, _ := json.Marshal(filter) |
| 217 | + sum := sha256.Sum256(raw) |
| 218 | + return "grasp02.history." + target + "." + hex.EncodeToString(sum[:]) |
| 219 | +} |
| 220 | + |
| 221 | +func (t *Tenant) gitHistoryWindow(ctx context.Context, target string, filter event.Filter) (event.Filter, bool, error) { |
| 222 | + var cursor storage.EventCursor |
| 223 | + err := t.store.GetSetting(ctx, gitHistoryCursorKey(target, filter), &cursor) |
| 224 | + if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 225 | + return filter, false, err |
| 226 | + } |
| 227 | + if cursor.CreatedAt > 0 { |
| 228 | + until := cursor.CreatedAt |
| 229 | + filter.Until = &until |
| 230 | + return filter, true, nil |
| 231 | + } |
| 232 | + return filter, false, nil |
| 233 | } |
| 234 | |
| 235 | func (t *Tenant) localSyncItems(ctx context.Context, filter event.Filter) ([]syncprotocol.Item, error) { |
| 236 | - rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10001}) |
| 237 | + rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10000}) |
| 238 | if err != nil { |
| 239 | return nil, err |
| 240 | } |
| 241 | - if rows.More { |
| 242 | - return nil, replication.ErrPullIncomplete |
| 243 | - } |
| 244 | + // A bounded inventory is intentionally reported as incomplete. The caller |
| 245 | + // switches to a bounded newest-first pull so a large local history cannot |
| 246 | + // overflow negentropy's missing-ID fetch cap. |
| 247 | items := make([]syncprotocol.Item, 0, len(rows.Events)) |
| 248 | for _, item := range rows.Events { |
| 249 | items = append(items, syncprotocol.Item{ID: item.ID, Timestamp: item.CreatedAt}) |
| 250 | } |
| 251 | + if rows.More { |
| 252 | + return items, replication.ErrPullIncomplete |
| 253 | + } |
| 254 | return items, nil |
| 255 | } |
| 256 | |
| 257 | @@ -167,9 +324,6 @@ func (t *Tenant) gitConversationEvents(ctx context.Context, repo gitrelay.Reposi |
| 258 | if err != nil { |
| 259 | return nil, err |
| 260 | } |
| 261 | - if rows.More { |
| 262 | - return nil, replication.ErrPullIncomplete |
| 263 | - } |
| 264 | items = append(items, rows.Events...) |
| 265 | } |
| 266 | roots := uniqueEvents(items) |
| 267 | @@ -178,9 +332,6 @@ func (t *Tenant) gitConversationEvents(ctx context.Context, repo gitrelay.Reposi |
| 268 | if err != nil { |
| 269 | return nil, err |
| 270 | } |
| 271 | - if rows.More { |
| 272 | - return nil, replication.ErrPullIncomplete |
| 273 | - } |
| 274 | items = append(items, rows.Events...) |
| 275 | } |
| 276 | return uniqueEvents(items), nil |
| 277 | diff --git a/internal/daemon/git_sync_cursor_test.go b/internal/daemon/git_sync_cursor_test.go |
| 278 | new file mode 100644 |
| 279 | index 0000000..f991e8a |
| 280 | --- /dev/null |
| 281 | +++ b/internal/daemon/git_sync_cursor_test.go |
| 282 | @@ -0,0 +1,217 @@ |
| 283 | +package daemon |
| 284 | + |
| 285 | +import ( |
| 286 | + "context" |
| 287 | + "encoding/json" |
| 288 | + "errors" |
| 289 | + "fmt" |
| 290 | + "net/http" |
| 291 | + "sort" |
| 292 | + "testing" |
| 293 | + "time" |
| 294 | + |
| 295 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 296 | + "github.com/FelineStateMachine/tinyrelay/internal/gitrelay" |
| 297 | + "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 298 | + "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 299 | +) |
| 300 | + |
| 301 | +// cursorRelay is deliberately a small NIP-01 relay. It applies the filter's |
| 302 | +// limit and until values, which makes the test exercise the same bounded head |
| 303 | +// and paginated older queries used by production sync. |
| 304 | +type cursorRelay struct { |
| 305 | + events []event.Event |
| 306 | + stallHistory bool |
| 307 | +} |
| 308 | + |
| 309 | +type cursorSocket struct { |
| 310 | + relay *cursorRelay |
| 311 | + responses [][]byte |
| 312 | +} |
| 313 | + |
| 314 | +type cursorDialer struct{ relay *cursorRelay } |
| 315 | + |
| 316 | +func (d cursorDialer) Dial(context.Context, string) (replication.Socket, *http.Response, error) { |
| 317 | + return &cursorSocket{relay: d.relay}, nil, nil |
| 318 | +} |
| 319 | + |
| 320 | +func (s *cursorSocket) Read(ctx context.Context) ([]byte, error) { |
| 321 | + if len(s.responses) == 0 { |
| 322 | + <-ctx.Done() |
| 323 | + return nil, ctx.Err() |
| 324 | + } |
| 325 | + item := s.responses[0] |
| 326 | + s.responses = s.responses[1:] |
| 327 | + return item, nil |
| 328 | +} |
| 329 | + |
| 330 | +func (s *cursorSocket) Write(_ context.Context, data []byte) error { |
| 331 | + var raw []json.RawMessage |
| 332 | + if err := json.Unmarshal(data, &raw); err != nil || len(raw) < 3 { |
| 333 | + return nil |
| 334 | + } |
| 335 | + var kind, sub string |
| 336 | + if json.Unmarshal(raw[0], &kind) != nil || json.Unmarshal(raw[1], &sub) != nil || kind != "REQ" { |
| 337 | + return nil |
| 338 | + } |
| 339 | + f, err := event.ParseFilter(raw[2]) |
| 340 | + if err != nil { |
| 341 | + return err |
| 342 | + } |
| 343 | + if f.Until != nil && s.relay.stallHistory { |
| 344 | + return nil |
| 345 | + } |
| 346 | + items := append([]event.Event(nil), s.relay.events...) |
| 347 | + sort.Slice(items, func(i, j int) bool { |
| 348 | + if items[i].CreatedAt != items[j].CreatedAt { |
| 349 | + return items[i].CreatedAt > items[j].CreatedAt |
| 350 | + } |
| 351 | + return items[i].ID < items[j].ID |
| 352 | + }) |
| 353 | + limit := 0 |
| 354 | + if f.Limit != nil { |
| 355 | + limit = *f.Limit |
| 356 | + } |
| 357 | + for _, item := range items { |
| 358 | + if !event.Matches(f, item) { |
| 359 | + continue |
| 360 | + } |
| 361 | + if limit > 0 && len(s.responses) >= limit { |
| 362 | + break |
| 363 | + } |
| 364 | + message, _ := json.Marshal([]any{"EVENT", sub, item}) |
| 365 | + s.responses = append(s.responses, message) |
| 366 | + } |
| 367 | + eose, _ := json.Marshal([]any{"EOSE", sub}) |
| 368 | + s.responses = append(s.responses, eose) |
| 369 | + return nil |
| 370 | +} |
| 371 | + |
| 372 | +func (s *cursorSocket) Close(error) error { return nil } |
| 373 | + |
| 374 | +func TestGitPullFilterAdvancesBeyondBoundedLocalInventory(t *testing.T) { |
| 375 | + _, tenant := testTenant(t) |
| 376 | + owner := tenant.Policy().Owner |
| 377 | + secret := "0000000000000000000000000000000000000000000000000000000000000001" |
| 378 | + now := time.Now().Unix() |
| 379 | + filter := event.Filter{Authors: []string{owner}, Kinds: []int{1}} |
| 380 | + for i := 0; i < 10001; i++ { |
| 381 | + // Seed the stored inventory directly; remote responses below still |
| 382 | + // undergo real signature validation and the normal ingestion path. |
| 383 | + item := event.Event{ID: fmt.Sprintf("%064x", i+1), PubKey: owner, Kind: 1, CreatedAt: now - int64(i+10000), Tags: [][]string{}, Content: "local"} |
| 384 | + if _, err := tenant.store.Save(context.Background(), item, storage.SaveOptions{Now: now, SearchMode: tenant.Policy().Features.Search}); err != nil { |
| 385 | + t.Fatal(err) |
| 386 | + } |
| 387 | + } |
| 388 | + remote := &cursorRelay{} |
| 389 | + for i := 0; i < gitHeadPageSize+1; i++ { |
| 390 | + item := event.Event{Kind: 1, CreatedAt: now - int64(i+1), Tags: [][]string{}, Content: "head"} |
| 391 | + if err := event.Sign(&item, secret); err != nil { |
| 392 | + t.Fatal(err) |
| 393 | + } |
| 394 | + remote.events = append(remote.events, item) |
| 395 | + } |
| 396 | + older := event.Event{Kind: 1, CreatedAt: now - 1000, Tags: [][]string{}, Content: "older"} |
| 397 | + if err := event.Sign(&older, secret); err != nil { |
| 398 | + t.Fatal(err) |
| 399 | + } |
| 400 | + remote.events = append(remote.events, older) |
| 401 | + transport := &replication.NostrTransport{Dialer: cursorDialer{relay: remote}, Timeout: time.Second} |
| 402 | + |
| 403 | + if err := tenant.gitPullFilter(context.Background(), transport, "ws://cursor.test", filter); !errors.Is(err, replication.ErrPullIncomplete) { |
| 404 | + t.Fatalf("first pull error = %v, want incomplete", err) |
| 405 | + } |
| 406 | + var cursor storage.EventCursor |
| 407 | + if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &cursor); err != nil || cursor.CreatedAt == 0 { |
| 408 | + t.Fatalf("head cursor was not persisted: %v %#v", err, cursor) |
| 409 | + } |
| 410 | + if got := countContent(t, tenant, "older"); got != 0 { |
| 411 | + t.Fatalf("older event arrived during head seed: %d", got) |
| 412 | + } |
| 413 | + if err := tenant.gitPullFilter(context.Background(), transport, "ws://cursor.test", filter); !errors.Is(err, replication.ErrPullIncomplete) { |
| 414 | + t.Fatalf("second pull error = %v, want incomplete local inventory", err) |
| 415 | + } |
| 416 | + if got := countContent(t, tenant, "older"); got != 1 { |
| 417 | + t.Fatalf("older event count = %d, want 1", got) |
| 418 | + } |
| 419 | + |
| 420 | + if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &cursor); err != nil { |
| 421 | + t.Fatal(err) |
| 422 | + } |
| 423 | + before := cursor |
| 424 | + ctx, cancel := context.WithCancel(context.Background()) |
| 425 | + cancel() |
| 426 | + if err := tenant.gitPullFilter(ctx, transport, "ws://cursor.test", filter); err == nil { |
| 427 | + t.Fatal("canceled pull returned nil") |
| 428 | + } |
| 429 | + var after storage.EventCursor |
| 430 | + if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &after); err != nil { |
| 431 | + t.Fatal(err) |
| 432 | + } |
| 433 | + if after != before { |
| 434 | + t.Fatalf("canceled pull advanced cursor from %#v to %#v", before, after) |
| 435 | + } |
| 436 | + // A stalled historical window must not consume the head or move its |
| 437 | + // checkpoint, even when the outer pass has a short deadline. |
| 438 | + before = storage.EventCursor{CreatedAt: older.CreatedAt, ID: older.ID} |
| 439 | + if err := tenant.store.PutSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), before); err != nil { |
| 440 | + t.Fatal(err) |
| 441 | + } |
| 442 | + newest := event.Event{Kind: 1, CreatedAt: now, Tags: [][]string{}, Content: "live-head"} |
| 443 | + if err := event.Sign(&newest, secret); err != nil { |
| 444 | + t.Fatal(err) |
| 445 | + } |
| 446 | + remote.events = []event.Event{newest} |
| 447 | + remote.stallHistory = true |
| 448 | + transport.Timeout = time.Minute |
| 449 | + ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second) |
| 450 | + defer cancel() |
| 451 | + if err := tenant.gitPullFilter(ctx, transport, "ws://cursor.test", filter); !errors.Is(err, context.DeadlineExceeded) { |
| 452 | + t.Fatalf("stalled history error = %v", err) |
| 453 | + } |
| 454 | + if got := countContent(t, tenant, "live-head"); got != 1 { |
| 455 | + t.Fatalf("stalled history blocked newest event: %d", got) |
| 456 | + } |
| 457 | + if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &after); err != nil { |
| 458 | + t.Fatal(err) |
| 459 | + } |
| 460 | + if after != before { |
| 461 | + t.Fatalf("stalled history moved cursor from %#v to %#v", before, after) |
| 462 | + } |
| 463 | +} |
| 464 | + |
| 465 | +func countContent(t *testing.T, tenant *Tenant, content string) int { |
| 466 | + t.Helper() |
| 467 | + rows, err := tenant.store.Query(context.Background(), event.Filter{Kinds: []int{1}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}}) |
| 468 | + if err != nil { |
| 469 | + t.Fatal(err) |
| 470 | + } |
| 471 | + count := 0 |
| 472 | + for _, item := range rows.Events { |
| 473 | + if item.Content == content { |
| 474 | + count++ |
| 475 | + } |
| 476 | + } |
| 477 | + return count |
| 478 | +} |
| 479 | + |
| 480 | +func TestGitSyncOutcomeClassifiesIncompleteAndBudgetDeadlines(t *testing.T) { |
| 481 | + ctx := context.Background() |
| 482 | + if got := gitSyncOutcome(ctx, errors.Join(context.DeadlineExceeded, replication.ErrPullIncomplete)); !errors.Is(got, gitrelay.ErrIncomplete) || errors.Is(got, replication.ErrPullIncomplete) { |
| 483 | + t.Fatalf("incomplete pass should map to gitrelay.ErrIncomplete, got %v", got) |
| 484 | + } |
| 485 | + if got := gitSyncOutcome(ctx, errors.Join(context.DeadlineExceeded, errors.New("dial failed"))); got == nil || errors.Is(got, gitrelay.ErrIncomplete) || errors.Is(got, context.DeadlineExceeded) { |
| 486 | + t.Fatalf("real errors must survive with deadlines dropped, got %v", got) |
| 487 | + } |
| 488 | + if got := gitSyncOutcome(ctx, context.DeadlineExceeded); got != nil { |
| 489 | + t.Fatalf("a bare budget deadline is not a failure, got %v", got) |
| 490 | + } |
| 491 | + canceled, cancel := context.WithCancel(ctx) |
| 492 | + cancel() |
| 493 | + if got := gitSyncOutcome(canceled, context.DeadlineExceeded); got == nil { |
| 494 | + t.Fatal("a canceled caller context must propagate") |
| 495 | + } |
| 496 | + if !gitIngestRejected(errors.New("blocked: this pubkey is banned")) || gitIngestRejected(errors.New("check pubkey ban: disk I/O error")) { |
| 497 | + t.Fatal("rejection classification must follow NIP-01 prefixes") |
| 498 | + } |
| 499 | +} |
| 500 | diff --git a/internal/daemon/git_sync_integration_test.go b/internal/daemon/git_sync_integration_test.go |
| 501 | index 8bfae17..c3f077c 100644 |
| 502 | --- a/internal/daemon/git_sync_integration_test.go |
| 503 | +++ b/internal/daemon/git_sync_integration_test.go |
| 504 | @@ -2,9 +2,16 @@ package daemon |
| 505 | |
| 506 | import ( |
| 507 | "context" |
| 508 | + "errors" |
| 509 | + "fmt" |
| 510 | + "strings" |
| 511 | "testing" |
| 512 | + "time" |
| 513 | |
| 514 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 515 | "github.com/FelineStateMachine/tinyrelay/internal/gitrelay" |
| 516 | + "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 517 | + "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 518 | ) |
| 519 | |
| 520 | func TestGitTargetTransportRequiresFeatureAndPrivatePeer(t *testing.T) { |
| 521 | @@ -23,3 +30,25 @@ func TestGitTargetTransportRequiresFeatureAndPrivatePeer(t *testing.T) { |
| 522 | t.Fatal("private repository could send filters to an unconfigured peer") |
| 523 | } |
| 524 | } |
| 525 | + |
| 526 | +func TestLocalSyncItemsUsesBoundedInventoryWindow(t *testing.T) { |
| 527 | + _, tenant := testTenant(t) |
| 528 | + owner := tenant.Policy().Owner |
| 529 | + now := time.Now().Unix() |
| 530 | + for i := 0; i < 10001; i++ { |
| 531 | + item := event.Event{ID: fmt.Sprintf("%064x", i+1), PubKey: owner, CreatedAt: now - int64(i), Kind: 1, Tags: [][]string{}, Content: "sync window"} |
| 532 | + if _, err := tenant.store.Save(context.Background(), item, storage.SaveOptions{Now: now, SearchMode: tenant.Policy().Features.Search}); err != nil { |
| 533 | + t.Fatalf("save event %d: %v", i, err) |
| 534 | + } |
| 535 | + } |
| 536 | + items, err := tenant.localSyncItems(context.Background(), event.Filter{Authors: []string{owner}, Kinds: []int{1}}) |
| 537 | + if !errors.Is(err, replication.ErrPullIncomplete) { |
| 538 | + t.Fatalf("inventory error = %v, want ErrPullIncomplete", err) |
| 539 | + } |
| 540 | + if len(items) != 10000 { |
| 541 | + t.Fatalf("inventory size = %d, want bounded 10000", len(items)) |
| 542 | + } |
| 543 | + if strings.TrimSpace(items[0].ID) == "" { |
| 544 | + t.Fatal("bounded inventory returned an empty event ID") |
| 545 | + } |
| 546 | +} |
| 547 | diff --git a/internal/daemon/tenant.go b/internal/daemon/tenant.go |
| 548 | index 24abbcd..3deb80a 100644 |
| 549 | --- a/internal/daemon/tenant.go |
| 550 | +++ b/internal/daemon/tenant.go |
| 551 | @@ -64,6 +64,7 @@ type Tenant struct { |
| 552 | policyWrite sync.Mutex |
| 553 | policy policy.Policy |
| 554 | maintenance maintenanceGate |
| 555 | + gitLegacy *replication.LegacyCache |
| 556 | } |
| 557 | |
| 558 | func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) { |
| 559 | @@ -82,7 +83,7 @@ func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) { |
| 560 | return nil, fmt.Errorf("preserve legacy repository privacy: %w", err) |
| 561 | } |
| 562 | } |
| 563 | - t := &Tenant{app: cfg.app, meta: cfg.meta, store: cfg.store, policy: p, publicURL: cfg.publicURL, auth: auth.NewValidator(time.Now), schedulerWake: make(chan struct{}, 1)} |
| 564 | + t := &Tenant{app: cfg.app, meta: cfg.meta, store: cfg.store, policy: p, publicURL: cfg.publicURL, auth: auth.NewValidator(time.Now), schedulerWake: make(chan struct{}, 1), gitLegacy: &replication.LegacyCache{}} |
| 565 | t.community, err = community.New(ctx, t.store, t.policy.Owner) |
| 566 | if err != nil { |
| 567 | return nil, err |
| 568 | diff --git a/internal/gitrelay/gitrelay.go b/internal/gitrelay/gitrelay.go |
| 569 | index 5a01864..a06d268 100644 |
| 570 | --- a/internal/gitrelay/gitrelay.go |
| 571 | +++ b/internal/gitrelay/gitrelay.go |
| 572 | @@ -282,7 +282,13 @@ func (s *Service) Tick(ctx context.Context) error { |
| 573 | } |
| 574 | cancel() |
| 575 | } |
| 576 | - if repoErr != nil { |
| 577 | + if errors.Is(repoErr, ErrIncomplete) { |
| 578 | + // Progress is saved; continue soon without recording a failure. |
| 579 | + if previous, ok := previousProgress.Repos[repoKey]; ok { |
| 580 | + status.SucceededAt = previous.SucceededAt |
| 581 | + } |
| 582 | + status.NextAt = progress.At + graspRetryDelay |
| 583 | + } else if repoErr != nil { |
| 584 | status.Error = repoErr.Error() |
| 585 | status.NextAt = progress.At + graspRetryDelay |
| 586 | tickErr = errors.Join(tickErr, fmt.Errorf("%s synchronization: %w", last, repoErr)) |
| 587 | @@ -924,6 +930,11 @@ func (g *GitRelay) validateRefs(refs map[string]string) error { |
| 588 | return nil |
| 589 | } |
| 590 | |
| 591 | +// ErrIncomplete reports a synchronization pass that saved progress and has |
| 592 | +// more history to fetch. The scheduler continues it soon without counting a |
| 593 | +// failure. |
| 594 | +var ErrIncomplete = errors.New("synchronization incomplete") |
| 595 | + |
| 596 | // Capabilities returns only protocol surfaces that this instance can serve. |
| 597 | func (g *GitRelay) Capabilities() []string { |
| 598 | p := g.policy() |
| 599 | diff --git a/internal/replication/websocket.go b/internal/replication/websocket.go |
| 600 | index d11d067..d1425f9 100644 |
| 601 | --- a/internal/replication/websocket.go |
| 602 | +++ b/internal/replication/websocket.go |
| 603 | @@ -256,6 +256,49 @@ type NostrTransport struct { |
| 604 | sequence atomic.Uint64 |
| 605 | legacyMu sync.Mutex |
| 606 | legacy map[string]struct{} |
| 607 | + // LegacyCache may be shared by short-lived transports created for the same |
| 608 | + // tenant. Entries expire so a relay can gain NIP-77 support without a |
| 609 | + // process restart. |
| 610 | + LegacyCache *LegacyCache |
| 611 | +} |
| 612 | + |
| 613 | +const legacyProbeTTL = 4 * time.Hour |
| 614 | + |
| 615 | +type LegacyCache struct { |
| 616 | + mu sync.Mutex |
| 617 | + entries map[string]time.Time |
| 618 | +} |
| 619 | + |
| 620 | +func (c *LegacyCache) known(target string) bool { |
| 621 | + c.mu.Lock() |
| 622 | + defer c.mu.Unlock() |
| 623 | + when, ok := c.entries[target] |
| 624 | + if !ok || time.Since(when) >= legacyProbeTTL { |
| 625 | + if ok { |
| 626 | + delete(c.entries, target) |
| 627 | + } |
| 628 | + return false |
| 629 | + } |
| 630 | + return true |
| 631 | +} |
| 632 | + |
| 633 | +func (c *LegacyCache) remember(target string) { |
| 634 | + c.mu.Lock() |
| 635 | + defer c.mu.Unlock() |
| 636 | + if c.entries == nil { |
| 637 | + c.entries = make(map[string]time.Time) |
| 638 | + } |
| 639 | + if len(c.entries) >= 64 { |
| 640 | + var oldest string |
| 641 | + var oldestAt time.Time |
| 642 | + for key, at := range c.entries { |
| 643 | + if oldest == "" || at.Before(oldestAt) { |
| 644 | + oldest, oldestAt = key, at |
| 645 | + } |
| 646 | + } |
| 647 | + delete(c.entries, oldest) |
| 648 | + } |
| 649 | + c.entries[target] = time.Now() |
| 650 | } |
| 651 | |
| 652 | // Subscribe keeps a bounded NIP-01 subscription open until ctx is canceled. |
| 653 | @@ -384,6 +427,9 @@ func (t *NostrTransport) QuerySynchronized(ctx context.Context, target string, f |
| 654 | } |
| 655 | |
| 656 | func (t *NostrTransport) knownLegacy(target string) bool { |
| 657 | + if t.LegacyCache != nil && t.LegacyCache.known(target) { |
| 658 | + return true |
| 659 | + } |
| 660 | t.legacyMu.Lock() |
| 661 | defer t.legacyMu.Unlock() |
| 662 | _, ok := t.legacy[target] |
| 663 | @@ -391,6 +437,10 @@ func (t *NostrTransport) knownLegacy(target string) bool { |
| 664 | } |
| 665 | |
| 666 | func (t *NostrTransport) rememberLegacy(target string) { |
| 667 | + if t.LegacyCache != nil { |
| 668 | + t.LegacyCache.remember(target) |
| 669 | + return |
| 670 | + } |
| 671 | t.legacyMu.Lock() |
| 672 | defer t.legacyMu.Unlock() |
| 673 | if t.legacy == nil { |
| 674 | diff --git a/internal/replication/websocket_transport_test.go b/internal/replication/websocket_transport_test.go |
| 675 | index 631e4d1..12d9e79 100644 |
| 676 | --- a/internal/replication/websocket_transport_test.go |
| 677 | +++ b/internal/replication/websocket_transport_test.go |
| 678 | @@ -3,6 +3,7 @@ package replication |
| 679 | import ( |
| 680 | "context" |
| 681 | "encoding/json" |
| 682 | + "fmt" |
| 683 | "net/http" |
| 684 | "testing" |
| 685 | "time" |
| 686 | @@ -63,8 +64,9 @@ func TestQuerySynchronizedFallsBackAfterSilentLegacyNegentropyPeer(t *testing.T) |
| 687 | |
| 688 | func TestQuerySynchronizedCachesSilentLegacyPeerPerTransport(t *testing.T) { |
| 689 | dialer := &silentNegentropyDialer{} |
| 690 | - transport := &NostrTransport{Dialer: dialer, Timeout: 10 * time.Millisecond} |
| 691 | + cache := &LegacyCache{} |
| 692 | for i := 0; i < 2; i++ { |
| 693 | + transport := &NostrTransport{Dialer: dialer, Timeout: 10 * time.Millisecond, LegacyCache: cache} |
| 694 | if _, err := transport.QuerySynchronized(context.Background(), "ws://legacy.example", event.Filter{}, nil); err != nil { |
| 695 | t.Fatal(err) |
| 696 | } |
| 697 | @@ -85,3 +87,20 @@ func TestQueryNegentropyIDsReportsMissingRequestedEvents(t *testing.T) { |
| 698 | t.Fatalf("returned %d events for missing ID", len(items)) |
| 699 | } |
| 700 | } |
| 701 | + |
| 702 | +func TestLegacyCacheExpiresAndBoundsEntries(t *testing.T) { |
| 703 | + cache := &LegacyCache{entries: map[string]time.Time{ |
| 704 | + "ws://expired.example": time.Now().Add(-legacyProbeTTL - time.Second), |
| 705 | + }} |
| 706 | + if cache.known("ws://expired.example") { |
| 707 | + t.Fatal("expired legacy probe was reused") |
| 708 | + } |
| 709 | + for i := 0; i < 65; i++ { |
| 710 | + cache.remember(fmt.Sprintf("ws://relay-%d.example", i)) |
| 711 | + } |
| 712 | + cache.mu.Lock() |
| 713 | + defer cache.mu.Unlock() |
| 714 | + if len(cache.entries) != 64 { |
| 715 | + t.Fatalf("cache entries = %d, want 64", len(cache.entries)) |
| 716 | + } |
| 717 | +} |