diff --git a/features/changes_follower.go b/features/changes_follower.go index 3891b45f..cdd64a06 100644 --- a/features/changes_follower.go +++ b/features/changes_follower.go @@ -36,6 +36,12 @@ import ( // exceed it. const forever time.Duration = math.MaxInt64 +// seqMarkersCapacity is the maximum number of seq marker entries retained. +const seqMarkersCapacity = 200 + +// seqMarkersEvictionCount is the number of entries removed when capacity is reached. +const seqMarkersEvictionCount = seqMarkersCapacity / 10 + // Minimal client timeout set to 1 minute. const minClientTimeout time.Duration = time.Minute @@ -53,6 +59,23 @@ const baseDelay time.Duration = 100 * time.Millisecond // Once we reach this number of retries we'll be capping the backoff var expRetryGate int = int(math.Log(float64(LongpollTimeout/baseDelay)) / math.Log(2)) +// seqEntryType distinguishes between a row (change item) and a page (batch) entry +// in the seq markers list. +type seqEntryType int + +const ( + seqEntryRow seqEntryType = iota + seqEntryPage +) + +// seqEntry records a sequence marker along with whether it came from a change +// row or from a page boundary. seq may be nil (e.g. when seq_interval causes +// the server to return a null seq). +type seqEntry struct { + entryType seqEntryType + seq *string +} + // Mode are enums for changes follower's operation mode. type Mode int @@ -136,6 +159,8 @@ type ChangesFollower struct { running bool runLock sync.Mutex logger core.Logger + seqMarkers []seqEntry + seqMarkersLock sync.RWMutex } // ChangesItem is a wrapper structure around cloudantv1.ChangesResultItem @@ -251,6 +276,89 @@ func (cf *ChangesFollower) StartOneOff() (<-chan ChangesItem, error) { return cf.run(Finite) } +// GetLastSeqNewerThan returns the newest sequence ID that is safe to use as a +// checkpoint after the given persisted sequence ID. +// +// Use this after fully processing a ChangesResultItem to determine whether +// this ChangesFollower has observed a later safe checkpoint. This is useful +// for filtered or sparse changes feeds, where the feed can advance across +// pages even when no additional user-processable change rows are returned. +// +// The supplied sequence ID must be the Seq of a ChangesResultItem that your +// application has fully processed and already persisted. This method returns +// a newer sequence only when doing so does not advance past later change rows +// that might not yet have been processed by your application. +// +// Returns the supplied ID unchanged when: +// - the follower has not started yet, +// - the supplied ID was not seen by this ChangesFollower instance, or +// - no newer safe checkpoint is available. +// +// Returns an error if lastPersistedSeqID is empty. +func (cf *ChangesFollower) GetLastSeqNewerThan(lastPersistedSeqID string) (string, error) { + if lastPersistedSeqID == "" { + return "", core.SDKErrorf(nil, "the provided sequence ID cannot be null or empty", "changes-follower-invalid-seq", common.GetComponentInfo()) + } + cf.seqMarkersLock.RLock() + defer cf.seqMarkersLock.RUnlock() + if len(cf.seqMarkers) == 0 { + return lastPersistedSeqID, nil + } + return cf.lastSeqSince(lastPersistedSeqID), nil +} + +// lastSeqSince walks forward through the retained seq markers from the given +// lastPersistedSeqID, fast-forwarding through consecutive page entries to +// return the furthest safe last_seq without advancing past later change rows +// that might not yet have been processed. +// +// Returns lastPersistedSeqID unchanged if not found in the markers. +// Must be called with seqMarkersLock at least read-held. +func (cf *ChangesFollower) lastSeqSince(lastPersistedSeqID string) string { + found := false + result := lastPersistedSeqID + + for _, entry := range cf.seqMarkers { + if found { + if entry.entryType == seqEntryRow { + break + } + result = *entry.seq + } else if entry.seq != nil && *entry.seq == lastPersistedSeqID { + found = true + result = *entry.seq + } + } + + if found { + return result + } + return lastPersistedSeqID +} + +// updateSeqMarkers updates the seq markers list with entries from a completed page. +// +// Evicts the oldest entries if the list is at capacity, then appends a ROW +// entry for the last change item (if any) and a PAGE entry for the page's +// last_seq. +func (cf *ChangesFollower) updateSeqMarkers(results []cloudantv1.ChangesResultItem, lastSeq *string) { + cf.seqMarkersLock.Lock() + defer cf.seqMarkersLock.Unlock() + if len(cf.seqMarkers) >= seqMarkersCapacity { + cf.seqMarkers = cf.seqMarkers[seqMarkersEvictionCount:] + } + if len(results) > 0 { + cf.seqMarkers = append(cf.seqMarkers, seqEntry{ + entryType: seqEntryRow, + seq: results[len(results)-1].Seq, + }) + } + cf.seqMarkers = append(cf.seqMarkers, seqEntry{ + entryType: seqEntryPage, + seq: lastSeq, + }) +} + // Stop this ChangesFollower. func (cf *ChangesFollower) Stop() { cf.cancel() @@ -436,6 +544,9 @@ func (cf *ChangesFollower) getChangesBatch() chan changesItems { if cf.suppression == Timer { cf.successTimestamp = time.Now() } + + cf.updateSeqMarkers(result.Results, result.LastSeq) + changes <- changesItems{items: result.Results} if cf.mode == Finite && *result.Pending == 0 { return diff --git a/features/changes_follower_test.go b/features/changes_follower_test.go index b08e0b97..e4499ddb 100644 --- a/features/changes_follower_test.go +++ b/features/changes_follower_test.go @@ -26,6 +26,7 @@ import ( "github.com/IBM/go-sdk-core/v5/core" . "github.com/onsi/ginkgo" + . "github.com/onsi/ginkgo/extensions/table" . "github.com/onsi/gomega" "github.com/onsi/gomega/gmeasure" ) @@ -1150,3 +1151,368 @@ var _ = Describe(`ChangesFollower with context`, func() { Expect(item).To(Equal(cloudantv1.ChangesResultItem{})) }) }) + +// --------------------------------------------------------------------------- +// seqMarkers helpers +// --------------------------------------------------------------------------- + +// seqStr builds a seq string from an integer, e.g. seqStr(11) -> "11-aa". +func seqStr(n int) string { + return fmt.Sprintf("%d-aa", n) +} + +// makeTestRow builds a ChangesResultItem with the given seq (nil for null rows). +func makeTestRow(s *string) cloudantv1.ChangesResultItem { + return cloudantv1.ChangesResultItem{ + ID: core.StringPtr("doc"), + Changes: []cloudantv1.Change{}, + Seq: s, + } +} + +// testPageData holds the raw page fields used to populate seqMarkers. +type testPageData struct { + results []cloudantv1.ChangesResultItem + lastSeq string +} + +// testPageType is the factory for the 9 page types. +// +// Type 1: rows=[b, b+1], lastSeq=b+1 (last row == last_seq, no nulls) +// Type 2: rows=[b, b+1], lastSeq=b+2 (last row != last_seq, no nulls) +// Type 3: rows=[null, b+1], lastSeq=b+1 (leading null, last row == last_seq) +// Type 4: rows=[null, b+1], lastSeq=b+2 (leading null, last row != last_seq) +// Type 5: rows=[b, null], lastSeq=b+1 (trailing null last row) +// Type 6: rows=[b, null], lastSeq=b+2 (trailing null last row, last_seq beyond) +// Type 7: rows=[null, null], lastSeq=b+1 (all nulls) +// Type 8: rows=[null, null], lastSeq=b+2 (all nulls, last_seq beyond) +// Type 9: rows=[], lastSeq=b (empty page) +// +// What gets stored in seqMarkers per type: +// +// Types 1,3: ROW('(b+1)-aa'), PAGE('(b+1)-aa') +// Types 2,4: ROW('(b+1)-aa'), PAGE('(b+2)-aa') +// Types 5,7: ROW(nil), PAGE('(b+1)-aa') +// Types 6,8: ROW(nil), PAGE('(b+2)-aa') +// Type 9: PAGE('b-aa') (no ROW) +func testPageType(t, base int) testPageData { + s := func(n int) *string { v := seqStr(n); return &v } + switch t { + case 1: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(s(base)), makeTestRow(s(base + 1))}, lastSeq: seqStr(base + 1)} + case 2: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(s(base)), makeTestRow(s(base + 1))}, lastSeq: seqStr(base + 2)} + case 3: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(nil), makeTestRow(s(base + 1))}, lastSeq: seqStr(base + 1)} + case 4: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(nil), makeTestRow(s(base + 1))}, lastSeq: seqStr(base + 2)} + case 5: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(s(base)), makeTestRow(nil)}, lastSeq: seqStr(base + 1)} + case 6: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(s(base)), makeTestRow(nil)}, lastSeq: seqStr(base + 2)} + case 7: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(nil), makeTestRow(nil)}, lastSeq: seqStr(base + 1)} + case 8: + return testPageData{results: []cloudantv1.ChangesResultItem{makeTestRow(nil), makeTestRow(nil)}, lastSeq: seqStr(base + 2)} + case 9: + return testPageData{results: []cloudantv1.ChangesResultItem{}, lastSeq: seqStr(base)} + default: + panic(fmt.Sprintf("unknown page type: %d", t)) + } +} + +// populateTestFollower builds a ChangesFollower and fills its seqMarkers +// by calling the real updateSeqMarkers method for each page. +func populateTestFollower(pages []testPageData) *ChangesFollower { + cf := &ChangesFollower{} + for _, p := range pages { + ls := p.lastSeq + cf.updateSeqMarkers(p.results, &ls) + } + return cf +} + +// lastSeqSinceHelper populates a follower with pages and calls lastSeqSince. +func lastSeqSinceHelper(pages []testPageData, querySeq string) string { + return populateTestFollower(pages).lastSeqSince(querySeq) +} + +// --------------------------------------------------------------------------- +// seqMarkers / lastSeqSince tests +// --------------------------------------------------------------------------- + +var _ = Describe(`seqMarkers / lastSeqSince`, func() { + + // ----------------------------------------------------------------------- + // Not-found / empty edge cases + // ----------------------------------------------------------------------- + + It(`testLastSeqSinceNotFound`, func() { + result := lastSeqSinceHelper([]testPageData{testPageType(1, 10)}, "999-ff") + Expect(result).To(Equal("999-ff")) + }) + + It(`testLastSeqSinceEmptySeqMarkers`, func() { + result := lastSeqSinceHelper([]testPageData{}, "1-aa") + Expect(result).To(Equal("1-aa")) + }) + + // ----------------------------------------------------------------------- + // Per-page-type: single page + // ----------------------------------------------------------------------- + + DescribeTable(`testLastSeqSinceAlone`, + func(pageT, base int, querySeq, expected string) { + result := lastSeqSinceHelper([]testPageData{testPageType(pageT, base)}, querySeq) + Expect(result).To(Equal(expected)) + }, + Entry("Type 1: last row seq (== last_seq)", 1, 10, seqStr(11), seqStr(11)), + Entry("Type 3: last row seq (== last_seq)", 3, 10, seqStr(11), seqStr(11)), + Entry("Type 2: last row seq -> last_seq", 2, 10, seqStr(11), seqStr(12)), + Entry("Type 2: last_seq key -> itself", 2, 10, seqStr(12), seqStr(12)), + Entry("Type 4: last row seq -> last_seq", 4, 10, seqStr(11), seqStr(12)), + Entry("Type 4: last_seq key -> itself", 4, 10, seqStr(12), seqStr(12)), + Entry("Type 5: non-stored row seq unchanged", 5, 10, seqStr(10), seqStr(10)), + Entry("Type 5: last_seq key -> itself", 5, 10, seqStr(11), seqStr(11)), + Entry("Type 6: non-stored row seq unchanged", 6, 10, seqStr(10), seqStr(10)), + Entry("Type 6: last_seq key -> itself", 6, 10, seqStr(12), seqStr(12)), + Entry("Type 7: last_seq key -> itself", 7, 10, seqStr(11), seqStr(11)), + Entry("Type 8: last_seq key -> itself", 8, 10, seqStr(12), seqStr(12)), + Entry("Type 9: last_seq key -> itself", 9, 10, seqStr(10), seqStr(10)), + ) + + // ----------------------------------------------------------------------- + // Per-page-type: followed by a non-empty page (type 1 at base 20) + // Page 2 inserts ROW('21-aa') which blocks advancement. + // ----------------------------------------------------------------------- + + DescribeTable(`testLastSeqSinceFollowedByNonEmpty`, + func(pageT, base int, querySeq, expected string) { + result := lastSeqSinceHelper([]testPageData{testPageType(pageT, base), testPageType(1, 20)}, querySeq) + Expect(result).To(Equal(expected)) + }, + Entry("Type 1 + non-empty: blocked by p2 ROW", 1, 10, seqStr(11), seqStr(11)), + Entry("Type 2 + non-empty: last row seq -> p1 last_seq", 2, 10, seqStr(11), seqStr(12)), + Entry("Type 2 + non-empty: last_seq key -> p1 last_seq", 2, 10, seqStr(12), seqStr(12)), + Entry("Type 3 + non-empty: blocked by p2 ROW", 3, 10, seqStr(11), seqStr(11)), + Entry("Type 4 + non-empty: last row seq -> p1 last_seq", 4, 10, seqStr(11), seqStr(12)), + Entry("Type 4 + non-empty: last_seq key -> p1 last_seq", 4, 10, seqStr(12), seqStr(12)), + Entry("Type 5 + non-empty: blocked by p2 ROW", 5, 10, seqStr(11), seqStr(11)), + Entry("Type 6 + non-empty: blocked by p2 ROW", 6, 10, seqStr(12), seqStr(12)), + Entry("Type 7 + non-empty: blocked by p2 ROW", 7, 10, seqStr(11), seqStr(11)), + Entry("Type 8 + non-empty: blocked by p2 ROW", 8, 10, seqStr(12), seqStr(12)), + Entry("Type 9 + non-empty: blocked by p2 ROW", 9, 10, seqStr(10), seqStr(10)), + ) + + // ----------------------------------------------------------------------- + // Per-page-type: followed by an empty page (type 9 at base 20) + // Page 2 inserts only PAGE('20-aa') — no ROW to block, advances to '20-aa'. + // ----------------------------------------------------------------------- + + DescribeTable(`testLastSeqSinceFollowedByEmpty`, + func(pageT, base int, querySeq, expected string) { + result := lastSeqSinceHelper([]testPageData{testPageType(pageT, base), testPageType(9, 20)}, querySeq) + Expect(result).To(Equal(expected)) + }, + Entry("Type 1 + empty: advances to p2 last_seq", 1, 10, seqStr(11), seqStr(20)), + Entry("Type 2 + empty: last row seq advances to p2", 2, 10, seqStr(11), seqStr(20)), + Entry("Type 2 + empty: last_seq key advances to p2", 2, 10, seqStr(12), seqStr(20)), + Entry("Type 3 + empty: advances to p2 last_seq", 3, 10, seqStr(11), seqStr(20)), + Entry("Type 4 + empty: last row seq advances to p2", 4, 10, seqStr(11), seqStr(20)), + Entry("Type 4 + empty: last_seq key advances to p2", 4, 10, seqStr(12), seqStr(20)), + Entry("Type 5 + empty: last_seq advances to p2", 5, 10, seqStr(11), seqStr(20)), + Entry("Type 6 + empty: last_seq advances to p2", 6, 10, seqStr(12), seqStr(20)), + Entry("Type 7 + empty: last_seq advances to p2", 7, 10, seqStr(11), seqStr(20)), + Entry("Type 8 + empty: last_seq advances to p2", 8, 10, seqStr(12), seqStr(20)), + Entry("Type 9 + empty: advances to p2 last_seq", 9, 10, seqStr(10), seqStr(20)), + ) + + // ----------------------------------------------------------------------- + // All 8 three-page sequences of empty (E=type 9) and non-empty (N=type 1). + // Query from page 1's last_seq key. E adds only PAGE; N adds ROW+PAGE. + // ----------------------------------------------------------------------- + + DescribeTable(`testLastSeqSince3PageSequence`, + func(types []int, bases []int, querySeq, expected string) { + pages := make([]testPageData, len(types)) + for i, t := range types { + pages[i] = testPageType(t, bases[i]) + } + result := lastSeqSinceHelper(pages, querySeq) + Expect(result).To(Equal(expected)) + }, + Entry("NNN: blocked by p2 ROW -> p1 last_seq", + []int{1, 1, 1}, []int{10, 20, 30}, seqStr(11), seqStr(11)), + Entry("NNE: blocked by p2 ROW -> p1 last_seq", + []int{1, 1, 9}, []int{10, 20, 30}, seqStr(11), seqStr(11)), + Entry("NEE: advances through both empty pages", + []int{1, 9, 9}, []int{10, 20, 30}, seqStr(11), seqStr(30)), + Entry("NEN: advances through p2 empty, stops at p3 ROW", + []int{1, 9, 1}, []int{10, 20, 30}, seqStr(11), seqStr(20)), + Entry("ENN: blocked by p2 ROW -> p1 last_seq", + []int{9, 1, 1}, []int{10, 20, 30}, seqStr(10), seqStr(10)), + Entry("ENE: blocked by p2 ROW -> p1 last_seq", + []int{9, 1, 9}, []int{10, 20, 30}, seqStr(10), seqStr(10)), + Entry("EEN: advances through p2, stops at p3 ROW", + []int{9, 9, 1}, []int{10, 20, 30}, seqStr(10), seqStr(20)), + Entry("EEE: advances through all three empty pages", + []int{9, 9, 9}, []int{10, 20, 30}, seqStr(10), seqStr(30)), + ) + + // ----------------------------------------------------------------------- + // Eviction + // + // Each non-empty page (type 2) adds 2 entries (ROW + PAGE). With + // CAPACITY=200 and EVICTION_COUNT=20, adding 101 pages triggers one + // eviction of the oldest 20 entries (first 10 pages). Entries for + // page 0 (base=0) should be gone; most recent should remain. + // ----------------------------------------------------------------------- + + It(`testLastSeqSinceEviction`, func() { + pages := make([]testPageData, 101) + for i := range pages { + pages[i] = testPageType(2, i*10) + } + cf := populateTestFollower(pages) + + // Page 0 (base=0): row=seqStr(1), page=seqStr(2) — evicted, returns input unchanged + Expect(cf.lastSeqSince(seqStr(1))).To(Equal(seqStr(1))) + Expect(cf.lastSeqSince(seqStr(2))).To(Equal(seqStr(2))) + + // Most recent page (base=1000): row=seqStr(1001), page=seqStr(1002) — still present + Expect(cf.lastSeqSince(seqStr(1001))).To(Equal(seqStr(1002))) + Expect(cf.lastSeqSince(seqStr(1002))).To(Equal(seqStr(1002))) + }) + + // ----------------------------------------------------------------------- + // Nil seq row (seq_interval scenario) — would panic without nil guard + // ----------------------------------------------------------------------- + + It(`testLastSeqSinceNilRowSeqDoesNotPanic`, func() { + // Types 5/6/7/8 store ROW(nil). Querying a seq not in the markers + // must return input unchanged without panicking. + pages := []testPageData{testPageType(5, 10)} + Expect(func() { + result := lastSeqSinceHelper(pages, seqStr(10)) + Expect(result).To(Equal(seqStr(10))) + }).NotTo(Panic()) + }) +}) + +// --------------------------------------------------------------------------- +// GetLastSeqNewerThan tests +// --------------------------------------------------------------------------- + +var _ = Describe(`GetLastSeqNewerThan`, func() { + var ( + glsnService *cloudantv1.CloudantV1 + glsnPostChangesOptions *cloudantv1.PostChangesOptions + ) + + BeforeEach(func() { + var serviceErr error + glsnService, serviceErr = cloudantv1.NewCloudantV1(&cloudantv1.CloudantV1Options{ + URL: "http://localhost:5984", + Authenticator: &core.NoAuthAuthenticator{}, + }) + Expect(serviceErr).ShouldNot(HaveOccurred()) + glsnPostChangesOptions = glsnService.NewPostChangesOptions("db") + }) + + It(`testGetLastSeqNewerThanWithEmptyString`, func() { + follower, err := NewChangesFollower(glsnService, glsnPostChangesOptions) + Expect(err).ShouldNot(HaveOccurred()) + + _, err = follower.GetLastSeqNewerThan("") + Expect(err).Should(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("the provided sequence ID cannot be null or empty")) + Expect(errors.As(err, &expectedErrType)).To(BeTrue()) + }) + + It(`testGetLastSeqNewerThanBeforeFeedStarts`, func() { + follower, err := NewChangesFollower(glsnService, glsnPostChangesOptions) + Expect(err).ShouldNot(HaveOccurred()) + + result, err := follower.GetLastSeqNewerThan("seq-a") + Expect(err).ShouldNot(HaveOccurred()) + Expect(result).To(Equal("seq-a")) + }) + + It(`testGetLastSeqNewerThanUnknownSeq`, func() { + ms := NewMockServer(1, noErrors) + svc := ms.Start() + defer ms.Stop() + + opts := svc.NewPostChangesOptions("db") + follower, err := NewChangesFollower(svc, opts) + Expect(err).ShouldNot(HaveOccurred()) + + ch, err := follower.StartOneOff() + Expect(err).ShouldNot(HaveOccurred()) + for range ch { + } + + result, err := follower.GetLastSeqNewerThan("seq-unknown") + Expect(err).ShouldNot(HaveOccurred()) + Expect(result).To(Equal("seq-unknown")) + }) + + It(`testGetLastSeqNewerThanMiddleOfBatch`, func() { + ms := NewMockServer(1, noErrors) + svc := ms.Start() + defer ms.Stop() + + opts := svc.NewPostChangesOptions("db") + follower, err := NewChangesFollower(svc, opts) + Expect(err).ShouldNot(HaveOccurred()) + + ch, err := follower.StartOneOff() + Expect(err).ShouldNot(HaveOccurred()) + + var items []cloudantv1.ChangesResultItem + for ci := range ch { + item, itemErr := ci.Item() + Expect(itemErr).ShouldNot(HaveOccurred()) + items = append(items, item) + } + Expect(len(items)).To(BeNumerically(">", 2)) + + // Only the last item's seq is stored — middle items are not in seqMarkers. + seqA := *items[0].Seq + seqB := *items[1].Seq + + resultA, err := follower.GetLastSeqNewerThan(seqA) + Expect(err).ShouldNot(HaveOccurred()) + Expect(resultA).To(Equal(seqA)) + + resultB, err := follower.GetLastSeqNewerThan(seqB) + Expect(err).ShouldNot(HaveOccurred()) + Expect(resultB).To(Equal(seqB)) + }) + + It(`testGetLastSeqNewerThanEndToEnd`, func() { + ms := NewMockServer(1, noErrors) + svc := ms.Start() + defer ms.Stop() + + opts := svc.NewPostChangesOptions("db") + follower, err := NewChangesFollower(svc, opts) + Expect(err).ShouldNot(HaveOccurred()) + + ch, err := follower.StartOneOff() + Expect(err).ShouldNot(HaveOccurred()) + + var lastItem cloudantv1.ChangesResultItem + for ci := range ch { + item, itemErr := ci.Item() + Expect(itemErr).ShouldNot(HaveOccurred()) + lastItem = item + } + + // The last item's seq is the stored row entry. MockChangesGenerator + // produces pages where last row seq == last_seq, so result equals input. + result, err := follower.GetLastSeqNewerThan(*lastItem.Seq) + Expect(err).ShouldNot(HaveOccurred()) + Expect(result).To(Equal(*lastItem.Seq)) + }) +})