You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2025/04/24 10:30:21 UTC
(couchdb) branch auto-delete-3 updated (d661c5525 -> a33e98c42)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch auto-delete-3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
from d661c5525 ensure subtype is in doc id
new f2cb0afe1 wip peer checkpoints for nouveau
new a33e98c42 consolidate peer id generation from sig into one place
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/couchdb/nouveau/api/IndexInfo.java | 11 ++++++++++-
.../java/org/apache/couchdb/nouveau/core/Index.java | 12 ++++++++++--
src/couch_mrview/src/couch_mrview_index.erl | 4 ++--
src/couch_mrview/src/couch_mrview_util.erl | 8 --------
src/dreyfus/src/dreyfus_index.erl | 12 ++----------
src/dreyfus/src/dreyfus_index_updater.erl | 2 +-
src/fabric/src/fabric_drop_seq.erl | 10 +++++++++-
src/nouveau/src/nouveau_index_updater.erl | 17 +++++++++++++++++
8 files changed, 51 insertions(+), 25 deletions(-)
(couchdb) 01/02: wip peer checkpoints for nouveau
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch auto-delete-3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit f2cb0afe17eacc2140c4012383c0d20cce4cb475
Author: Robert Newson <rn...@apache.org>
AuthorDate: Wed Apr 23 16:49:05 2025 +0100
wip peer checkpoints for nouveau
---
.../org/apache/couchdb/nouveau/api/IndexInfo.java | 11 +++++++++-
.../org/apache/couchdb/nouveau/core/Index.java | 12 +++++++++--
src/nouveau/src/nouveau_index_updater.erl | 24 ++++++++++++++++++++++
3 files changed, 44 insertions(+), 3 deletions(-)
diff --git a/nouveau/src/main/java/org/apache/couchdb/nouveau/api/IndexInfo.java b/nouveau/src/main/java/org/apache/couchdb/nouveau/api/IndexInfo.java
index 9958c7780..51fbb0129 100644
--- a/nouveau/src/main/java/org/apache/couchdb/nouveau/api/IndexInfo.java
+++ b/nouveau/src/main/java/org/apache/couchdb/nouveau/api/IndexInfo.java
@@ -24,6 +24,9 @@ public final class IndexInfo {
@PositiveOrZero
private final long updateSeq;
+ @PositiveOrZero
+ private final long committedUpdateSeq;
+
@PositiveOrZero
private final long purgeSeq;
@@ -35,10 +38,12 @@ public final class IndexInfo {
public IndexInfo(
@JsonProperty("update_seq") final long updateSeq,
+ @JsonProperty("committed_update_seq") final long committedUpdateSeq,
@JsonProperty("purge_seq") final long purgeSeq,
@JsonProperty("num_docs") final int numDocs,
@JsonProperty("disk_size") final long diskSize) {
this.updateSeq = updateSeq;
+ this.committedUpdateSeq = committedUpdateSeq;
this.purgeSeq = purgeSeq;
this.numDocs = numDocs;
this.diskSize = diskSize;
@@ -56,13 +61,17 @@ public final class IndexInfo {
return updateSeq;
}
+ public long getCommittedUpdateSeq() {
+ return committedUpdateSeq;
+ }
+
public long getPurgeSeq() {
return purgeSeq;
}
@Override
public String toString() {
- return "IndexInfo [updateSeq=" + updateSeq + ", purgeSeq=" + purgeSeq + ", numDocs=" + numDocs + ", diskSize="
+ return "IndexInfo [updateSeq=" + updateSeq + ", committedUpdateSeq=" + committedUpdateSeq + ", purgeSeq=" + purgeSeq + ", numDocs=" + numDocs + ", diskSize="
+ diskSize + "]";
}
}
diff --git a/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java b/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java
index 69c9bd48f..bcb5859e4 100644
--- a/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java
+++ b/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java
@@ -36,18 +36,20 @@ import org.apache.couchdb.nouveau.api.SearchResults;
public abstract class Index implements Closeable {
private long updateSeq;
+ private long committedUpdateSeq;
private long purgeSeq;
private boolean deleteOnClose = false;
protected Index(final long updateSeq, final long purgeSeq) {
this.updateSeq = updateSeq;
+ this.committedUpdateSeq = updateSeq;
this.purgeSeq = purgeSeq;
}
public final IndexInfo info() throws IOException {
final int numDocs = doNumDocs();
final long diskSize = doDiskSize();
- return new IndexInfo(updateSeq, purgeSeq, numDocs, diskSize);
+ return new IndexInfo(updateSeq, committedUpdateSeq, purgeSeq, numDocs, diskSize);
}
protected abstract int doNumDocs() throws IOException;
@@ -93,7 +95,13 @@ public abstract class Index implements Closeable {
updateSeq = this.updateSeq;
purgeSeq = this.purgeSeq;
}
- return doCommit(updateSeq, purgeSeq);
+ final boolean result = doCommit(updateSeq, purgeSeq);
+ if (result) {
+ synchronized (this) {
+ this.committedUpdateSeq = Math.max(this.committedUpdateSeq, updateSeq);
+ }
+ }
+ return result;
}
protected abstract boolean doCommit(final long updateSeq, final long purgeSeq) throws IOException;
diff --git a/src/nouveau/src/nouveau_index_updater.erl b/src/nouveau/src/nouveau_index_updater.erl
index efed245db..526725960 100644
--- a/src/nouveau/src/nouveau_index_updater.erl
+++ b/src/nouveau/src/nouveau_index_updater.erl
@@ -87,6 +87,13 @@ update(#index{} = Index) ->
{ok, PurgeAcc1} = purge_index(ConnPid, Db, Index, PurgeAcc0),
NewCurSeq = couch_db:get_update_seq(Db),
+ fabric_drop_seq:create_peer_checkpoint_doc_if_missing(
+ Index#index.dbname,
+ <<"nouveau">>,
+ <<(Index#index.ddoc_id)/binary, "/", (Index#index.name)/binary>>,
+ peer_checkpoint_id(Index#index.dbname, Index#index.sig),
+ NewCurSeq
+ ),
Proc = get_os_process(Index#index.def_lang),
try
true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def, <<"nouveau">>]),
@@ -107,6 +114,16 @@ update(#index{} = Index) ->
Db, Acc0#acc.update_seq, fun load_docs/2, Acc0, []
),
nouveau_api:drain_async_responses(Acc1#acc.reqids, 0),
+ {ok, #{<<"committed_update_seq">> := CommittedUpdateSeq}} = nouveau_api:index_info(
+ Index
+ ),
+ fabric_drop_seq:update_peer_checkpoint_doc(
+ Index#index.dbname,
+ <<"nouveau">>,
+ <<(Index#index.ddoc_id)/binary, "/", (Index#index.name)/binary>>,
+ peer_checkpoint_id(Index#index.dbname, Index#index.sig),
+ CommittedUpdateSeq
+ ),
exit(nouveau_api:set_update_seq(ConnPid, Index, Acc1#acc.update_seq, NewCurSeq))
after
ibrowse:stop_worker_process(ConnPid),
@@ -289,3 +306,10 @@ update_local_doc(Db, #index{} = Index, PurgeSeq) ->
DocId = nouveau_util:get_local_purge_doc_id(Index#index.sig),
DocContent = nouveau_util:get_local_purge_doc_body(DocId, PurgeSeq, Index),
couch_db:update_doc(Db, DocContent, []).
+
+peer_checkpoint_id(DbName, Sig0) ->
+ Sig1 = couch_util:encodeBase64Url(Sig0),
+ Hash = couch_util:encodeBase64Url(
+ crypto:hash(sha256, [atom_to_binary(node()), $0, DbName])
+ ),
+ <<Sig1/binary, "-", Hash/binary>>.
(couchdb) 02/02: consolidate peer id generation from sig into one place
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch auto-delete-3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit a33e98c424eac3dd48aa728aaf90c67fe7c4a54d
Author: Robert Newson <rn...@apache.org>
AuthorDate: Thu Apr 24 11:27:45 2025 +0100
consolidate peer id generation from sig into one place
---
src/couch_mrview/src/couch_mrview_index.erl | 4 ++--
src/couch_mrview/src/couch_mrview_util.erl | 8 --------
src/dreyfus/src/dreyfus_index.erl | 12 ++----------
src/dreyfus/src/dreyfus_index_updater.erl | 2 +-
src/fabric/src/fabric_drop_seq.erl | 10 +++++++++-
src/nouveau/src/nouveau_index_updater.erl | 11 ++---------
6 files changed, 16 insertions(+), 31 deletions(-)
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 2c755e6f7..b7d08bddb 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -222,7 +222,7 @@ commit(State) ->
State#mrst.db_name,
<<"mrview">>,
State#mrst.idx_name,
- couch_mrview_util:peer_checkpoint_id(State),
+ fabric_drop_seq:peer_id_from_sig(State#mrst.db_name, State#mrst.sig),
State#mrst.update_seq
),
ok.
@@ -315,7 +315,7 @@ ensure_peer_checkpoint_doc(#mrst{} = State) ->
State#mrst.db_name,
<<"mrview">>,
State#mrst.idx_name,
- couch_mrview_util:peer_checkpoint_id(State),
+ fabric_drop_seq:peer_id_from_sig(State#mrst.db_name, State#mrst.sig),
State#mrst.update_seq
).
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index f0e708722..a478685da 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -36,7 +36,6 @@
-export([get_collator_versions/1]).
-export([compact_on_collator_upgrade/0]).
-export([commit_on_header_upgrade/0]).
--export([peer_checkpoint_id/1]).
-define(MOD, couch_mrview_index).
-define(GET_VIEW_RETRY_COUNT, 1).
@@ -1356,10 +1355,3 @@ compact_on_collator_upgrade() ->
commit_on_header_upgrade() ->
config:get_boolean("view_upgrade", "commit_on_header_upgrade", true).
-
-peer_checkpoint_id(#mrst{} = State) ->
- Sig = couch_util:encodeBase64Url(State#mrst.sig),
- Hash = couch_util:encodeBase64Url(
- crypto:hash(sha256, [atom_to_binary(node()), $0, State#mrst.db_name])
- ),
- <<Sig/binary, "-", Hash/binary>>.
diff --git a/src/dreyfus/src/dreyfus_index.erl b/src/dreyfus/src/dreyfus_index.erl
index 717dd85bf..c2b6fa6ea 100644
--- a/src/dreyfus/src/dreyfus_index.erl
+++ b/src/dreyfus/src/dreyfus_index.erl
@@ -28,8 +28,7 @@
info/1,
group1/2,
group2/2,
- design_doc_to_indexes/1,
- peer_checkpoint_id/2
+ design_doc_to_indexes/1
]).
% gen_server api.
@@ -129,7 +128,7 @@ init({DbName, Index}) ->
DbName,
<<"search">>,
<<(Index#index.ddoc_id)/binary, "/", (Index#index.name)/binary>>,
- peer_checkpoint_id(DbName, Index#index.sig),
+ fabric_drop_seq:peer_id_from_sig(DbName, Index#index.sig),
Seq
),
proc_lib:init_ack({ok, self()}),
@@ -429,10 +428,3 @@ group2_int(Pid, QueryArgs0) ->
info_int(Pid) ->
clouseau_rpc:info(Pid).
-
-peer_checkpoint_id(DbName, Sig0) ->
- Sig1 = couch_util:encodeBase64Url(Sig0),
- Hash = couch_util:encodeBase64Url(
- crypto:hash(sha256, [atom_to_binary(node()), $0, DbName])
- ),
- <<Sig1/binary, "-", Hash/binary>>.
diff --git a/src/dreyfus/src/dreyfus_index_updater.erl b/src/dreyfus/src/dreyfus_index_updater.erl
index 178f34de4..f8404cc53 100644
--- a/src/dreyfus/src/dreyfus_index_updater.erl
+++ b/src/dreyfus/src/dreyfus_index_updater.erl
@@ -66,7 +66,7 @@ update(IndexPid, Index) ->
DbName,
<<"search">>,
<<(Index#index.ddoc_id)/binary, "/", (Index#index.name)/binary>>,
- dreyfus_index:peer_checkpoint_id(
+ fabric_drop_seq:peer_id_from_sig(
DbName,
Index#index.sig
),
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 5ad5c395e..65795c9ab 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -10,7 +10,8 @@
-export([
create_peer_checkpoint_doc_if_missing/5,
update_peer_checkpoint_doc/5,
- peer_checkpoint_doc/4
+ peer_checkpoint_doc/4,
+ peer_id_from_sig/2
]).
-type range() :: [non_neg_integer()].
@@ -369,6 +370,13 @@ peer_checkpoint_doc(PeerId, Subtype, Source, UpdateSeq) when
peer_checkpoint_id(Subtype, PeerId) ->
<<?LOCAL_DOC_PREFIX, "peer-checkpoint-", Subtype/binary, "-", PeerId/binary>>.
+peer_id_from_sig(DbName, Sig0) ->
+ Sig1 = couch_util:encodeBase64Url(Sig0),
+ Hash = couch_util:encodeBase64Url(
+ crypto:hash(sha256, [atom_to_binary(node()), $0, DbName])
+ ),
+ <<Sig1/binary, "-", Hash/binary>>.
+
pack_seq(DbName, UpdateSeq) ->
DbUuid = couch_util:with_db(DbName, fun(Db) -> couch_db:get_uuid(Db) end),
Seq0 = [node(), mem3:range(DbName), {UpdateSeq, DbUuid, node()}],
diff --git a/src/nouveau/src/nouveau_index_updater.erl b/src/nouveau/src/nouveau_index_updater.erl
index 526725960..6ba3c3738 100644
--- a/src/nouveau/src/nouveau_index_updater.erl
+++ b/src/nouveau/src/nouveau_index_updater.erl
@@ -91,7 +91,7 @@ update(#index{} = Index) ->
Index#index.dbname,
<<"nouveau">>,
<<(Index#index.ddoc_id)/binary, "/", (Index#index.name)/binary>>,
- peer_checkpoint_id(Index#index.dbname, Index#index.sig),
+ fabric_drop_seq:peer_id_from_sig(Index#index.dbname, Index#index.sig),
NewCurSeq
),
Proc = get_os_process(Index#index.def_lang),
@@ -121,7 +121,7 @@ update(#index{} = Index) ->
Index#index.dbname,
<<"nouveau">>,
<<(Index#index.ddoc_id)/binary, "/", (Index#index.name)/binary>>,
- peer_checkpoint_id(Index#index.dbname, Index#index.sig),
+ fabric_drop_seq:peer_id_from_sig(Index#index.dbname, Index#index.sig),
CommittedUpdateSeq
),
exit(nouveau_api:set_update_seq(ConnPid, Index, Acc1#acc.update_seq, NewCurSeq))
@@ -306,10 +306,3 @@ update_local_doc(Db, #index{} = Index, PurgeSeq) ->
DocId = nouveau_util:get_local_purge_doc_id(Index#index.sig),
DocContent = nouveau_util:get_local_purge_doc_body(DocId, PurgeSeq, Index),
couch_db:update_doc(Db, DocContent, []).
-
-peer_checkpoint_id(DbName, Sig0) ->
- Sig1 = couch_util:encodeBase64Url(Sig0),
- Hash = couch_util:encodeBase64Url(
- crypto:hash(sha256, [atom_to_binary(node()), $0, DbName])
- ),
- <<Sig1/binary, "-", Hash/binary>>.