- commit
- 7d30ccdf656cc6881f36e27661401c11b7ac51c0
- parent
- 74b3905e76bff61925afb9b7bb169d8034e90f69
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 13 | ++++++++--- |
| M | internal/ | 10 | ++++++--- |
| M | internal/ | 36 | ++++++++++++++++++++++++++++++ |
@@ -154,7 +154,9 @@ func (t *Tenant) gitSyncPeer(ctx context.Context, repo gitrelay.Repository, targ
154154 return err155155 }156156 var result error157+ cursors := make(map[string]struct{})157158 for _, filter := range gitEventFilters(repo) {159+ cursors[gitHistoryCursorKey(target, filter)] = struct{}{}158160 result = errors.Join(result, t.gitPullFilter(ctx, transport, target, filter))159161 }160162 events, err := t.gitConversationEvents(ctx, repo)@@ -162,9 +164,12 @@ func (t *Tenant) gitSyncPeer(ctx context.Context, repo gitrelay.Repository, targ
162164 return errors.Join(result, err)163165 }164166 for _, filter := range gitReplyFilters(events) {167+ cursors[gitHistoryCursorKey(target, filter)] = struct{}{}165168 result = errors.Join(result, t.gitPullFilter(ctx, transport, target, filter))166169 }167- return result170+ // Reply filters are chunked by root ID, so their keys shift as roots171+ // arrive. Drop cursors for filters this pass no longer uses.172+ return errors.Join(result, t.store.PruneSettings(ctx, gitHistoryCursorPrefix(target), cursors))168173 }169174 170175 func (t *Tenant) gitPullFilter(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) error {@@ -216,7 +221,7 @@ func (t *Tenant) gitPullLargeFilter(ctx context.Context, transport *replication.
216221 if olderErr == nil {217222 // Finished this sweep. The next pass starts another from the head,218223 // allowing older events newly acquired by the peer to be discovered.219- result = errors.Join(result, t.store.PutSetting(ctx, key, storage.EventCursor{}))224+ result = errors.Join(result, t.store.DeleteSetting(ctx, key))220225 } else if errors.Is(olderErr, replication.ErrPullIncomplete) && len(older) > 0 {221226 result = errors.Join(result, t.store.PutSetting(ctx, key, oldestGitHistoryCursor(older)))222227 }@@ -279,10 +284,12 @@ func oldestGitHistoryCursor(items []event.Event) storage.EventCursor {
279284 return storage.EventCursor{CreatedAt: oldest.CreatedAt, ID: oldest.ID}280285 }281286 287+func gitHistoryCursorPrefix(target string) string { return "grasp02.history." + target + "." }288+282289 func gitHistoryCursorKey(target string, filter event.Filter) string {283290 raw, _ := json.Marshal(filter)284291 sum := sha256.Sum256(raw)285- return "grasp02.history." + target + "." + hex.EncodeToString(sum[:])292+ return gitHistoryCursorPrefix(target) + hex.EncodeToString(sum[:])286293 }287294 288295 func (t *Tenant) gitHistoryWindow(ctx context.Context, target string, filter event.Filter) (event.Filter, bool, error) {@@ -2,6 +2,7 @@ package daemon
22 33 import (44 "context"5+ "database/sql"56 "encoding/json"67 "errors"78 "fmt"@@ -135,8 +136,11 @@ func TestGitPullFilterAdvancesBeyondBoundedLocalInventory(t *testing.T) {
135136 t.Fatalf("older event count = %d, want 1", got)136137 }137138 138- if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &cursor); err != nil {139- t.Fatal(err)139+ // A completed sweep removes its cursor row so per-filter state cannot140+ // accumulate; the next pass seeds a new head window.141+ cursor = storage.EventCursor{}142+ if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &cursor); !errors.Is(err, sql.ErrNoRows) {143+ t.Fatalf("completed sweep left cursor %#v, err %v", cursor, err)140144 }141145 before := cursor142146 ctx, cancel := context.WithCancel(context.Background())@@ -145,7 +149,7 @@ func TestGitPullFilterAdvancesBeyondBoundedLocalInventory(t *testing.T) {
145149 t.Fatal("canceled pull returned nil")146150 }147151 var after storage.EventCursor148- if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &after); err != nil {152+ if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &after); err != nil && !errors.Is(err, sql.ErrNoRows) {149153 t.Fatal(err)150154 }151155 if after != before {@@ -114,6 +114,42 @@ func (s *Store) PutSetting(ctx context.Context, key string, value any) error {
114114 return s.WithTx(ctx, func(tx *sql.Tx) error { return PutSetting(ctx, tx, key, value) })115115 }116116 117+// DeleteSetting removes one settings row. A missing row is not an error.118+func (s *Store) DeleteSetting(ctx context.Context, key string) error {119+ _, err := s.db.ExecContext(ctx, "DELETE FROM settings WHERE key=?", key)120+ return err121+}122+123+// PruneSettings removes rows under prefix whose keys are not in keep, so124+// per-filter state cannot accumulate as filters come and go.125+func (s *Store) PruneSettings(ctx context.Context, prefix string, keep map[string]struct{}) error {126+ rows, err := s.db.QueryContext(ctx, "SELECT key FROM settings WHERE substr(key,1,?)=?", len(prefix), prefix)127+ if err != nil {128+ return err129+ }130+ var stale []string131+ for rows.Next() {132+ var key string133+ if err := rows.Scan(&key); err != nil {134+ rows.Close()135+ return err136+ }137+ if _, ok := keep[key]; !ok {138+ stale = append(stale, key)139+ }140+ }141+ rows.Close()142+ if err := rows.Err(); err != nil {143+ return err144+ }145+ for _, key := range stale {146+ if err := s.DeleteSetting(ctx, key); err != nil {147+ return err148+ }149+ }150+ return nil151+}152+117153 func PutSetting(ctx context.Context, tx *sql.Tx, key string, value any) error {118154 raw, err := json.Marshal(value)119155 if err != nil {