Given multiple sequence numbers. How to find oldest?

[ronag]

I have some custom replication logic that shards the replication and each replication work hence has it's own saved sequence (checkpoint). In the case of a restart, the different workers might have saved different sequence checkpoints.

Given multiple sequence identifiers, is there a way to determine the "oldest" one. So that I can avoid having to start up separate change streams for each worker and instead just have one shared change stream that starts from the oldest sequence.

[nickva]

The number in the front of the sequence would a rough estimate. So if it's "47-g1A..." it would be the 47 part.

In general it not possible to easily compare those. If you're curious and have remsh access to the cluster you can try to introspect the opaque sequence using the fabric_view_changes:decode_seq/1 and see how it contains references to individual shards, integer sequences to those, node and shard uuids, etc.

Here is a one example:

(node1@127.0.0.1)2> rp(fabric_view_changes:decode_seq("47-g1AAAAoheJyVlT9M10AUxyuYsOlkotBNJwdz_3p3nXRzM1Ho5tL70xDCv25ujiQK3djYSBC6sbE5KHRzczNR6Obm5iD8fEcN11DTcP0NL-mv98l73_d975ajKHq0OG2iu6trxpJnmIgnCH54Gf6YWr0dJVX7G8IEnqXF6fzWCryfUQJzVODhqSsSHpJW4-YXBAZATypSU1CVD0-NkT7U9U8IKwB0pKlLksUWi1SFkX7IqoWwB0CfU4JlkVA5QmJD0oOm_AbhOwAdKfpH0pKrYnhqLKdXWfYVwn0AOtL0JYlTpHKEwqp738ovEF4C0FdHpOCC8jDS5zJuILwDoFdcWIElESMkOiSdTyYfIXwCoNdJyoQRiYanxnLiVXsE4Q8Ae70ziHJNw6pbi5vDzupHvrqUWeAEKr5f13Vn9cP-tCRIszRMp1NZ7XZWr31OihFlqArTabYpdzqr7_amBRWpRklYdfNZtt1ZfafXO4sEUTaMtNnKqrP6ttcplybP6dgu-M_cHZfxRmf1yufEhKYFUwFzp95OJhed0Td6eluapIUJqE29aSvhYOeekhBuEhWitSqbeN2huK_JUC6IGFPnuovUUl0fONSap6SSMZvbAAcpXckzh9r3lAITrIOmTL0umzmHOvU7TcJ2TDkLoWRZtuBQs55SKCSoHNvWA11eyHbLoea991KhLL6a0Rvp8jwuTxxq03eaIJknDIdU9BQuVoc69hVpxCjnQa6TbSUd6qI3TZZiIsfm8vo0KdzEpUMJn0vOLU0THjJJj2FlOdR6bwMWzBYq5M5RDyvZOtSBd51KCo4ICaHE4H6HOut1mimKr-7kG-lyD7ziUHO-0yYnKE1YiC53QFmHWvAV5RpzrYMqmoE8HGrL98hgi7gam4DrlDxynBOfCDfaMK6Hny_9BYUIh7I")).
[{'[email protected]',[4175662645,4294967295],
                    {1,<<"b7160f1">>,'[email protected]'}},
 {'[email protected]',[4056357998,4175662644],
                    {1,<<"f9df3ba">>,'[email protected]'}},
 {'[email protected]',[3937053351,4056357997],
                    {2,<<"e1e179b">>,'[email protected]'}},
 {'[email protected]',[3817748704,3937053350],
                    {1,<<"518f538">>,'[email protected]'}},
 {'[email protected]',[3698444057,3817748703],
                    {0,<<"51c86bf">>,'[email protected]'}},
 {'[email protected]',[3579139410,3698444056],
                    {3,<<"630ba00">>,'[email protected]'}},
 {'[email protected]',[3459834763,3579139409],
                    {1,<<"2876736">>,'[email protected]'}},
 {'[email protected]',[3340530116,3459834762],
                    {2,<<"7e71827">>,'[email protected]'}},
 {'[email protected]',[3221225469,3340530115],
                    {0,<<"8854280">>,'[email protected]'}},
 {'[email protected]',[3101920822,3221225468],
                    {1,<<"5d036c3">>,'[email protected]'}},
 {'[email protected]',[2982616175,3101920821],
                    {2,<<"94ec3d0">>,'[email protected]'}},
 {'[email protected]',[2863311528,2982616174],
                    {1,<<"b750c49">>,'[email protected]'}},
 {'[email protected]',[2744006881,2863311527],
                    {2,<<"b42bd3b">>,'[email protected]'}},
 {'[email protected]',[2624702234,2744006880],
                    {2,<<"e0f9c05">>,'[email protected]'}},
 {'[email protected]',[2505397587,2624702233],
                    {0,<<"8e072be">>,'[email protected]'}},
 {'[email protected]',[2386092940,2505397586],
                    {1,<<"a8daa3a">>,'[email protected]'}},
 {'[email protected]',[2266788293,2386092939],
                    {0,<<"47c3f4b">>,'[email protected]'}},
 {'[email protected]',[2147483646,2266788292],
                    {2,<<"be359fd">>,'[email protected]'}},
 {'[email protected]',[2028178999,2147483645],
                    {2,<<"526d5b5">>,'[email protected]'}},
 {'[email protected]',[1908874352,2028178998],
                    {0,<<"d36727a">>,'[email protected]'}},
 {'[email protected]',[1789569705,1908874351],
                    {0,<<"9844eae">>,'[email protected]'}},
 {'[email protected]',[1670265058,1789569704],
                    {0,<<"f121cc3">>,'[email protected]'}},
 {'[email protected]',[1550960411,1670265057],
                    {3,<<"81c8964">>,'[email protected]'}},
 {'[email protected]',[1431655764,1550960410],
                    {3,<<"fb07388">>,'[email protected]'}},
 {'[email protected]',[1312351117,1431655763],
                    {1,<<"97be149">>,'[email protected]'}},
 {'[email protected]',[1193046470,1312351116],
                    {2,<<"208a541">>,'[email protected]'}},
 {'[email protected]',[1073741823,1193046469],
                    {3,<<"c04366d">>,'[email protected]'}},
 {'[email protected]',[954437176,1073741822],
                    {1,<<"ae3128e">>,'[email protected]'}},
 {'[email protected]',[835132529,954437175],
                    {3,<<"a6e3956">>,'[email protected]'}},
 {'[email protected]',[715827882,835132528],
                    {1,<<"bf4efb6">>,'[email protected]'}},
 {'[email protected]',[596523235,715827881],
                    {0,<<"b5f6022">>,'[email protected]'}},
 {'[email protected]',[477218588,596523234],
                    {1,<<"94b31bf">>,'[email protected]'}},
 {'[email protected]',[357913941,477218587],
                    {2,<<"da20954">>,'[email protected]'}},
 {'[email protected]',[238609294,357913940],
                    {0,<<"ac16cc2">>,'[email protected]'}},
 {'[email protected]',[119304647,238609293],
                    {3,<<"d1e06b8">>,'[email protected]'}},
 {'[email protected]',[0,119304646],
                    {0,<<"6dcd46c">>,'[email protected]'}}]

Also if you're curious to learn more about the changes feed, I like this resource https://blog.cloudant.com/2022/01/21/Using-the-Cloudant-changes-feed.html

[ronag]

I actually don't need an exact sequence just a sequence that is guaranteed to be before any of the listed sequences. I don't think checking the front of the sequence is sufficient in a cluster i.e. there are several different sums of sequence numbers that can provide the same number.

I think the sequence string is built something like this:

{sum of all shard seq}-${shard1_seq}-{shard2_seq} etc... so I kind of need to be able to parse, update and re-encode this.

Is there any documentation of how to do this?

[ronag]

As I currently understand the format:

{'[email protected]',[119304647,238609293], {3,<<"d1e06b8">>,'[email protected]'}}

'[email protected]', is the node
[119304647,238609293] is some kind of range. Not sure exactly what this represents?
3 is the shard sequence number
<<"d1e06b8">> I have no idea what it is?
'[email protected]' I have no idea why it occurs again here?

[ronag]

Basically what I want to achieve is something like:

for await (const change of couch.changes({
  since: getOldestSequence(workers.map(({ last_seq }) => last_seq)) // On restart safely resume
})) { 
  for (const worker of workers) {
    worker.send(change)
  }
}

In order to avoid having each worker (which can be 64+) having their own separate change stream.

[nickva]

To avoid each worker having to checkpoint just have one worker checkpoint or have them all delegate to some common component to checkpoint. Then next time start since=... checkpointed sequence. Otherwise each worker should have its own sequence and checkpoint it separately, as they are not strictly comparable.

You can do a rough comparison based on the prefix N-... (which is just sum of the total changes across all the shards which responded first, it's best not use that f you don't expect to miss any changes.

I showed what they contain internally and mostly for learning purposes. The range is the shards range, the 7 character uuid is part of the uuid of the shard which responded. The node name is repeated because the second on is the epoch, that is the node which owns that particular sequence range, as technically shards could be moved to other nodes during their lifetime.

[ronag]

To avoid each worker having to checkpoint just have one worker checkpoint or have them all delegate to some common component to checkpoint.

This won't work as I can't guarantee that each worker has the same checkpoint given e.g. a process crash. Basically what each worker does is:

const last_seq = await db.get('last_seq') // resume in case of restart or crash
for await (const changes of couch.changes({ since: last_seq, batch: true }))
  await db.write([
    ...getRecords(changes),
    { last_seq: changes.at(-1).seq }
  ])
}

That way if I have a crash I can ensure that the data written to the worker db is consistent.

Otherwise each worker should have its own sequence and checkpoint it separately, as they are not strictly comparable.

Which isn't working either. On a 128 core machine with 256 hardware threads I can have up to 256 workers that all need a change stream. Which is a significant overhead (and something I've noticed couch just can't handle) vs just having 1 change stream which broadcasts to all workers using a shared ring buffer which has very low overheard...

I showed what they contain internally and mostly for learning purposes.

Unfortunately, I don't see any other way to solve this other than decoding the sequences and generating a new one that is guaranteed to include all changes for every worker..

[ronag]

An alternative would be if couch could take an array of sequence numbers as the since parameter and guarantee to include the oldest in the resulting change stream? That way this logic does not need to be performed in "user space".

[nickva]

An alternative would be if couch could take an array of sequence numbers as the since parameter and guarantee to include the oldest in the resulting change stream? That way this logic does not need to be performed in "user space".

That could work but only when comparing the same shard copies (matching shard uuids exactly). In general shard copies for the same range are not guaranteed to have been built in the same order (so seq=10 on db1 and db2 could be different updates). So naively comparing sequences against each doesn't work well. But I think it may be possible if we query all the shard copies and also consider the internal replication checkpoints between the copies. So it would be a bit tricky to do right.

[ronag]

Tricky but possible? What would it require to try and have this done? Either inside or outside of couch.

[ronag]

Maybe another alternative would be:

Grab now. Make sure all workers are update "until" "now" and then restart the change feed from that sequence. But then I need the change stream to support a "until" parameter, e.g.

const start_seq = await changesWorker.getStartSeq()
const last_seq = await db.get('last_seq') // resume in case of restart or crash
for await (const changes of couch.changes({ since: start_seq, until: last_seq, batch: true }))
  await db.write([
    ...getRecords(changes),
    { last_seq: changes.at(-1).seq }
  ])
}

await changesWorker.signalReady()

for await (const changes of changesWorker.changes({ batch: true }))
  await db.write([
    ...getRecords(changes),
    { last_seq: changes.at(-1).seq }
  ])
}

[ronag]

Is there a way I can emulate until in user land?

[nickva]

Unfortunately there is no until in CouchDB. It might be interesting to have perhaps at some point. I've wanted an "until" parameter at some point as well in the past.