ffi: JobSubject and ContentKey cross as themselves

Kotlin named a job's subject through JobSubjectRef and an artifact through
DocumentAssetRef — FFI-only copies of JobSubject and ContentKey, each with
a TryFrom back. They existed because the real enums had tuple variants,
which uniffi hands Kotlin as v1/v2. Both now have named fields (Peer
{ node }, DocumentToPeer { document_id, node }, Blob { hash }, Artifact
{ id }, …) and derive uniffi::Enum, so the copies and their conversions
go. BlobHash crosses as its "blake3:<hex>" string and a bare Uuid as its
hyphenated one; a malformed one from Kotlin lifts as InvalidInput.

job_match_key now takes the subject a job row is built from, so the
drift-guard test that compared the two spellings has nothing left to
compare. (F4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Greg Shuflin
2026-10-03 17:31:56 -07:00
co-authored by Claude Opus 5.5
parent dcd8ed77a6
commit ebb276b77e
47 changed files with 348 additions and 422 deletions
@@ -46,8 +46,8 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import uniffi.synchronicity.ChangeChannel
import uniffi.synchronicity.ContentKey
import uniffi.synchronicity.DocumentArtifactItem
import uniffi.synchronicity.DocumentAssetRef
import uniffi.synchronicity.DocumentBlobRefItem
import uniffi.synchronicity.DocumentItem
import uniffi.synchronicity.DocumentNodeStatusItem
@@ -57,7 +57,7 @@ import uniffi.synchronicity.DocumentPropertyItem
import uniffi.synchronicity.DocumentSyncOverride
import uniffi.synchronicity.DocumentSyncOverrideOption
import uniffi.synchronicity.JobKind
import uniffi.synchronicity.JobSubjectRef
import uniffi.synchronicity.JobSubject
import uniffi.synchronicity.assetSyncPushed
import uniffi.synchronicity.assetSyncStopped
import uniffi.synchronicity.assetSyncSummary
@@ -109,7 +109,7 @@ fun DocumentDetailScreen(documentId: String, onBack: () -> Unit, modifier: Modif
mutableStateOf<List<DocumentSyncOverrideOption>>(emptyList())
}
// Per-artifact pull state for the Artifacts section's own "Sync" buttons, keyed by the
// artifact (see [DocumentAssetRef.key]) rather than by list position: a successful fetch
// artifact (see [ContentKey.key]) rather than by list position: a successful fetch
// bumps `reloadKey` and rebuilds the rows, so an index would follow whatever row landed in
// that slot instead of the one the user tapped.
val assetSyncStates = remember { mutableStateMapOf<String, AssetSyncState>() }
@@ -187,7 +187,7 @@ fun DocumentDetailScreen(documentId: String, onBack: () -> Unit, modifier: Modif
// "Sync" button. The narrow counterpart to the Sync tab's whole-document control: this
// fetches exactly `asset` and doesn't recurse into child documents. Several artifacts can
// be in flight at once, each reporting its own outcome.
fun syncAsset(asset: DocumentAssetRef) {
fun syncAsset(asset: ContentKey) {
val key = asset.key
if (assetSyncStates[key]?.starting == true) return
assetSyncStates[key] = AssetSyncState(starting = true)
@@ -672,8 +672,8 @@ private fun PlacementNodeRow(
val control =
rememberJobControl(
if (row.isThisNode) JobKind.CONTENT_FETCH else JobKind.PUSH,
if (row.isThisNode) JobSubjectRef.Document(documentId)
else JobSubjectRef.DocumentToPeer(documentId = documentId, nodeIdHex = row.nodeIdHex),
if (row.isThisNode) JobSubject.Document(documentId)
else JobSubject.DocumentToPeer(documentId = documentId, node = row.nodeIdHex),
starting = actionState.starting,
)
Column(
@@ -847,9 +847,9 @@ private fun DangerZoneCard(
// The eviction itself is a job, so it reports whoever started it — this screen before it was
// backgrounded, or the Sync policy screen's type-scoped "Reclaim…" — and survives leaving and
// re-entering. `Evaluating` stays this screen's: gathering the evidence for the dialog is not
// work anyone else can see. `JobSubjectRef.Document`, not `DocumentType`: this is one subtree.
// work anyone else can see. `JobSubject.Document`, not `DocumentType`: this is one subtree.
val evicting =
rememberJobControl(JobKind.EVICTION, JobSubjectRef.Document(documentId = documentId))
rememberJobControl(JobKind.EVICTION, JobSubject.Document(documentId = documentId))
DetailCard("Danger Zone") {
Row(
modifier = Modifier.fillMaxWidth(),
@@ -999,11 +999,11 @@ private fun KindAndSummary(kind: String, modifiedAt: String, summary: String) {
* Stable map key for one artifact's [AssetSyncState]. The two id spaces can't collide: a UUID never
* looks like `"blake3:…"`.
*/
private val DocumentAssetRef.key: String
private val ContentKey.key: String
get() =
when (this) {
is DocumentAssetRef.Blob -> hash
is DocumentAssetRef.Artifact -> id
is ContentKey.Blob -> hash
is ContentKey.Artifact -> id
}
/**
@@ -1033,7 +1033,7 @@ private fun ArtifactsCard(
artifacts: List<DocumentArtifactItem>,
blobRefs: List<DocumentBlobRefItem>,
syncStates: Map<String, AssetSyncState>,
onSync: (DocumentAssetRef) -> Unit,
onSync: (ContentKey) -> Unit,
) {
DetailCard("Artifacts", trailing = "${artifacts.size + blobRefs.size}") {
if (artifacts.isEmpty() && blobRefs.isEmpty()) {
@@ -1045,7 +1045,7 @@ private fun ArtifactsCard(
} else {
blobRefs.forEachIndexed { index, blobRef ->
if (index > 0) HorizontalDivider(thickness = 0.5.dp)
val asset = DocumentAssetRef.Blob(blobRef.blobHash)
val asset = ContentKey.Blob(blobRef.blobHash)
BlobRefRow(
documentId = documentId,
blobRef = blobRef,
@@ -1058,7 +1058,7 @@ private fun ArtifactsCard(
if (index > 0 || blobRefs.isNotEmpty()) {
HorizontalDivider(thickness = 0.5.dp)
}
val asset = DocumentAssetRef.Artifact(artifact.id)
val asset = ContentKey.Artifact(artifact.id)
ArtifactRow(
documentId = documentId,
artifact = artifact,
@@ -1075,7 +1075,7 @@ private fun ArtifactsCard(
private fun ArtifactRow(
documentId: String,
artifact: DocumentArtifactItem,
asset: DocumentAssetRef,
asset: ContentKey,
syncState: AssetSyncState,
onSync: () -> Unit,
) {
@@ -1106,7 +1106,7 @@ private fun ArtifactRow(
@Composable
private fun AssetSyncControl(
documentId: String,
asset: DocumentAssetRef,
asset: ContentKey,
presentLocally: Boolean,
state: AssetSyncState,
onSync: () -> Unit,
@@ -1114,7 +1114,7 @@ private fun AssetSyncControl(
val control =
rememberJobControl(
JobKind.CONTENT_FETCH,
JobSubjectRef.DocumentAsset(documentId = documentId, asset = asset),
JobSubject.DocumentAsset(documentId = documentId, asset = asset),
starting = state.starting,
)
Row(
@@ -1149,7 +1149,7 @@ private fun AssetSyncControl(
private fun BlobRefRow(
documentId: String,
blobRef: DocumentBlobRefItem,
asset: DocumentAssetRef,
asset: ContentKey,
syncState: AssetSyncState,
onSync: () -> Unit,
) {
@@ -19,7 +19,7 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import uniffi.synchronicity.JobKind
import uniffi.synchronicity.JobSubjectRef
import uniffi.synchronicity.JobSubject
import uniffi.synchronicity.formatSeenTimestamp
import uniffi.synchronicity.pingTimeoutSecs
@@ -74,7 +74,7 @@ internal fun SyncControl(nodeIdHex: String, state: SyncState, onSync: () -> Unit
val control =
rememberJobControl(
JobKind.CATALOG_SYNC,
JobSubjectRef.Peer(nodeIdHex = nodeIdHex),
JobSubject.Peer(node = nodeIdHex),
starting = state.starting,
)
val pending = control.busy
@@ -21,7 +21,7 @@ import com.gregshuflin.synchronicity.ffiCatching
import com.gregshuflin.synchronicity.ui.InlineSpinner
import uniffi.synchronicity.JobItem
import uniffi.synchronicity.JobKind
import uniffi.synchronicity.JobSubjectRef
import uniffi.synchronicity.JobSubject
import uniffi.synchronicity.jobActivityText
import uniffi.synchronicity.jobMatchKey
@@ -67,11 +67,7 @@ data class JobControl(val job: JobItem?, val starting: Boolean = false) {
* @param starting whether this screen has just started this work and is waiting for its job row.
*/
@Composable
fun rememberJobControl(
kind: JobKind,
subject: JobSubjectRef,
starting: Boolean = false,
): JobControl {
fun rememberJobControl(kind: JobKind, subject: JobSubject, starting: Boolean = false): JobControl {
val matchKey = rememberJobMatchKey(kind, subject)
val snapshot by ActiveJobs.snapshot.collectAsState()
return JobControl(snapshot.jobFor(matchKey), starting)
@@ -83,7 +79,7 @@ fun rememberJobControl(
* caller bug and is logged as one.
*/
@Composable
private fun rememberJobMatchKey(kind: JobKind, subject: JobSubjectRef): String? =
private fun rememberJobMatchKey(kind: JobKind, subject: JobSubject): String? =
remember(kind, subject) {
ffiCatching { jobMatchKey(kind, subject) }
.onFailure { Log.e(TAG, "Can't build a job key for $kind/$subject", it) }
@@ -48,7 +48,7 @@ import kotlinx.coroutines.launch
import uniffi.synchronicity.ChangeChannel
import uniffi.synchronicity.DocumentType
import uniffi.synchronicity.JobKind
import uniffi.synchronicity.JobSubjectRef
import uniffi.synchronicity.JobSubject
import uniffi.synchronicity.LibraryEntry
import uniffi.synchronicity.LibraryEntryKind
import uniffi.synchronicity.LibraryTreeItem
@@ -372,7 +372,7 @@ private fun RebuildLibraryAction(starting: Boolean, onRebuild: () -> Unit) {
val control =
rememberJobControl(
JobKind.LIBRARY_BUILD,
JobSubjectRef.DocumentType(DocumentType.MUSIC_TRACK),
JobSubject.DocumentType(DocumentType.MUSIC_TRACK),
starting = starting,
)
if (control.busy) {
@@ -20,7 +20,7 @@ import uniffi.synchronicity.ContentRow
import uniffi.synchronicity.DocumentSyncPolicy
import uniffi.synchronicity.DocumentType
import uniffi.synchronicity.JobKind
import uniffi.synchronicity.JobSubjectRef
import uniffi.synchronicity.JobSubject
import uniffi.synchronicity.RemoteDocumentSyncPolicy
import uniffi.synchronicity.SyncPolicy
import uniffi.synchronicity.SyncPolicyOption
@@ -62,9 +62,9 @@ fun SyncPolicyControl(
// A running eviction for this type outranks whatever this screen's own flow last said, and is
// reported even when there is no flow at all: it may have been started from the document
// detail screen's "Delete Locally", or here before the screen was last backgrounded.
// `JobSubjectRef.DocumentType` rather than `Document`, so a type-scoped reclaim dedups against
// `JobSubject.DocumentType` rather than `Document`, so a type-scoped reclaim dedups against
// itself rather than colliding with a per-document eviction of something in that type.
val reclaiming = rememberJobControl(JobKind.EVICTION, JobSubjectRef.DocumentType(entry.docType))
val reclaiming = rememberJobControl(JobKind.EVICTION, JobSubject.DocumentType(entry.docType))
Column(modifier = modifier.fillMaxWidth(), verticalArrangement = Arrangement.spacedBy(4.dp)) {
PolicyPicker(entry.displayName, entry.policy, entry.content, options) {
onPolicyChange(entry.docType, it)
@@ -48,7 +48,7 @@ import uniffi.synchronicity.DocumentSyncPolicy
import uniffi.synchronicity.DocumentType
import uniffi.synchronicity.EnsembleMemberInfo
import uniffi.synchronicity.JobKind
import uniffi.synchronicity.JobSubjectRef
import uniffi.synchronicity.JobSubject
import uniffi.synchronicity.RemoteDocumentSyncPolicy
import uniffi.synchronicity.SchedulerSettings
import uniffi.synchronicity.getSyncPolicyOptions
@@ -416,11 +416,11 @@ private fun BackgroundSyncSection(
onRunNow: () -> Unit,
) {
// The job decides, not this screen: a pass may already be under way because the periodic tick
// fired or a peer sent `TriggerSync`. `JobSubjectRef.None` because a full pass is about the
// fired or a peer sent `TriggerSync`. `JobSubject.None` because a full pass is about the
// node, not one peer or document — which is also what makes those three dedup against each
// other.
val control =
rememberJobControl(JobKind.BACKGROUND_SYNC, JobSubjectRef.None, starting = runNowStarting)
rememberJobControl(JobKind.BACKGROUND_SYNC, JobSubject.None, starting = runNowStarting)
Column(verticalArrangement = Arrangement.spacedBy(8.dp)) {
SectionHeading("Background sync")
ExplainerText(
+5 -4
View File
@@ -388,10 +388,11 @@ either frontend.
one that starts it inside a spawned task (the Ensemble screen's "Sync")
passes `starting: true` for the gap.
`JobSubjectRef` is the FFI spelling of the subject, and it carries
`DocumentType` as the enum rather than its snake_case name for the reason
`job_match_key` exists at all: a frontend that had to produce that name would
be a mirror of `DocumentType::type_name`, and mirrors go stale silently.
`JobKind` and `JobSubject` cross the FFI as themselves, so a widget names its
work with the same values the job row was made from — `DocumentType` as the
enum rather than its snake_case name, for the reason `job_match_key` exists at
all: a frontend that had to produce that name would be a mirror of
`DocumentType::type_name`, and mirrors go stale silently.
What stays a per-screen flag on both is the *result* of the last action —
"Pushed 3 documents", "Sync failed: …". A job carries a state and a progress
@@ -428,7 +428,7 @@ impl DocumentsView {
let doc_id = detail.id.clone();
let Some(handle) = self.job_runner.start(
JobKind::ContentFetch,
JobSubject::Document(doc_id.clone()),
JobSubject::Document { document_id: doc_id.clone() },
format!("Pull {doc_id}"),
JobOrigin::User,
) else {
@@ -515,7 +515,7 @@ impl DocumentsView {
// whole-document pull of the same document.
let Some(handle) = self.job_runner.start(
JobKind::ContentFetch,
JobSubject::DocumentAsset(doc_id.clone(), asset.clone()),
JobSubject::DocumentAsset { document_id: doc_id.clone(), asset: asset.clone() },
format!("Pull {asset} of {doc_id}"),
JobOrigin::User,
) else {
@@ -592,9 +592,15 @@ impl DocumentsView {
/// this screen's to remember.
fn row_job_control(&self, detail: &Detail, row: &PlacementRow) -> JobControl {
let (kind, subject) = if row.is_this_node {
(JobKind::ContentFetch, JobSubject::Document(detail.id.clone()))
(JobKind::ContentFetch, JobSubject::Document { document_id: detail.id.clone() })
} else {
(JobKind::Push, JobSubject::DocumentToPeer(detail.id.clone(), row.node_pubkey))
(
JobKind::Push,
JobSubject::DocumentToPeer {
document_id: detail.id.clone(),
node: row.node_pubkey,
},
)
};
self.active_jobs.control(kind, &subject, false)
}
@@ -606,7 +612,8 @@ impl DocumentsView {
/// ([`JobSubject::DocumentAsset`] against [`JobSubject::Document`]), and
/// one must not make the other's control look busy.
fn item_job_control(&self, detail: &Detail, asset: &ContentKey) -> JobControl {
let subject = JobSubject::DocumentAsset(detail.id.clone(), asset.clone());
let subject =
JobSubject::DocumentAsset { document_id: detail.id.clone(), asset: asset.clone() };
self.active_jobs.control(JobKind::ContentFetch, &subject, false)
}
@@ -616,7 +623,7 @@ impl DocumentsView {
/// subtree, and it must not dedup against the Settings screen's
/// type-scoped "Reclaim…" — nor be masked by it.
fn delete_locally_control(&self, detail: &Detail) -> JobControl {
let subject = JobSubject::Document(detail.id.clone());
let subject = JobSubject::Document { document_id: detail.id.clone() };
self.active_jobs.control(JobKind::Eviction, &subject, false)
}
@@ -660,7 +667,7 @@ impl DocumentsView {
let node_pubkey = row.node_pubkey;
let Some(handle) = self.job_runner.start(
JobKind::Push,
JobSubject::DocumentToPeer(doc_id.clone(), node_pubkey),
JobSubject::DocumentToPeer { document_id: doc_id.clone(), node: node_pubkey },
format!("Push {doc_id} to {}", node_pubkey.to_hex()),
JobOrigin::User,
) else {
@@ -952,7 +959,7 @@ impl DocumentsView {
let Some(handle) = self.job_runner.start(
JobKind::Eviction,
JobSubject::Document(doc_id.clone()),
JobSubject::Document { document_id: doc_id.clone() },
format!("Delete locally: {doc_id}"),
JobOrigin::User,
) else {
@@ -226,7 +226,7 @@ fn artifact_rows(
summary: artifact_value::summary(&artifact.value, VALUE_PREVIEW_CHARS).into(),
content_hash: None,
node_statuses,
asset: ContentKey::Artifact(artifact.id),
asset: ContentKey::Artifact { id: artifact.id },
// These are the rows built from real values, not stubs.
present_locally: true,
}
@@ -264,7 +264,7 @@ fn artifact_rows(
summary: summary.into(),
content_hash: None,
node_statuses,
asset: ContentKey::Artifact(stub.id),
asset: ContentKey::Artifact { id: stub.id },
present_locally: false,
});
}
@@ -292,7 +292,7 @@ fn blob_rows(
summary: size.into(),
content_hash: Some(blob_ref.blob_hash.to_string().into()),
node_statuses,
asset: ContentKey::Blob(blob_ref.blob_hash.clone()),
asset: ContentKey::Blob { hash: blob_ref.blob_hash.clone() },
// A blob row always exists once the pointer has synced; the
// ciphertext is a separate question, and the only one that
// decides whether there is anything left to fetch.
@@ -349,7 +349,7 @@ fn artifact_node_statuses(
// Either `artifact` or `stub` identifies the row; both carry the artifact
// id the evidence store is keyed on.
let evidence_key =
artifact.map(|a| a.id).or_else(|| stub.map(|s| s.id)).map(ContentKey::Artifact);
artifact.map(|a| a.id).or_else(|| stub.map(|s| s.id)).map(|id| ContentKey::Artifact { id });
let evidence_for = |node: &NodePubkey| {
let key = evidence_key.as_ref()?;
store
@@ -418,7 +418,7 @@ fn blob_node_statuses(
peer_names: &HashMap<String, String>,
tz: Option<chrono_tz::Tz>,
) -> Vec<NodeStatus> {
let key = ContentKey::Blob(blob_ref.blob_hash.clone());
let key = ContentKey::Blob { hash: blob_ref.blob_hash.clone() };
let mut rows: Vec<_> = store
.node_content_status_store
.batch(|tx| {
+2 -1
View File
@@ -672,7 +672,8 @@ impl DocumentsView {
.into_iter()
.filter(|job| job.kind == JobKind::ContentFetch)
.filter_map(|job| match job.subject {
JobSubject::Document(id) | JobSubject::DocumentAsset(id, _) => Some(id.to_string()),
JobSubject::Document { document_id: id }
| JobSubject::DocumentAsset { document_id: id, .. } => Some(id.to_string()),
_ => None,
})
.collect();
+2 -1
View File
@@ -365,7 +365,8 @@ impl JobsView {
/// with at job-creation time. Every other kind still shows
/// `job.kind.display_name()` + `job.label` as plain text.
fn render_label_row(&self, theme: &Theme, job: &Job) -> gpui::AnyElement {
if let (JobKind::Push, JobSubject::DocumentToPeer(doc_id, peer)) = (job.kind, &job.subject)
if let (JobKind::Push, JobSubject::DocumentToPeer { document_id: doc_id, node: peer }) =
(job.kind, &job.subject)
{
let doc_id_str = doc_id.to_string();
let title = self.doc_titles.get(&doc_id_str).cloned().map(SharedString::from);
+2 -2
View File
@@ -786,7 +786,7 @@ impl MusicView {
/// The live `LibraryBuild` job for this agent's type, whoever started it.
fn library_build_control(&self) -> JobControl {
let subject = JobSubject::DocumentType(DocumentType::MusicTrack);
let subject = JobSubject::DocumentType { doc_type: DocumentType::MusicTrack };
self.active_jobs.control(JobKind::LibraryBuild, &subject, false)
}
@@ -805,7 +805,7 @@ impl MusicView {
// frame's answer, and another surface can start one between the two.
let Some(handle) = self.job_runner.start(
JobKind::LibraryBuild,
JobSubject::DocumentType(DocumentType::MusicTrack),
JobSubject::DocumentType { doc_type: DocumentType::MusicTrack },
"Build music library",
JobOrigin::User,
) else {
@@ -42,7 +42,9 @@ pub(super) fn save<V: 'static>(
let handle = job_runner
.start(
JobKind::Import,
JobSubject::Document(podcast::episode_document_id(&feed_id, &episode.api_id)),
JobSubject::Document {
document_id: podcast::episode_document_id(&feed_id, &episode.api_id),
},
episode_download::job_label(&episode.title),
JobOrigin::User,
)
+1 -1
View File
@@ -139,7 +139,7 @@ impl RemoteFetch {
/// blank between the job finishing and the content being decoded.
pub fn refresh(&mut self, runner: &JobRunner) {
let Some(document) = self.document.clone() else { return };
let subject = JobSubject::Document(document);
let subject = JobSubject::Document { document_id: document };
let Some(job) = runner
.list_active()
+6 -2
View File
@@ -38,8 +38,12 @@ pub fn start(
error: &mut Option<SharedString>,
) -> Option<JobHandle> {
let label = label.into();
let handle =
job_runner.start(JobKind::Import, JobSubject::Path(label.clone()), label, JobOrigin::User);
let handle = job_runner.start(
JobKind::Import,
JobSubject::Path { path: label.clone() },
label,
JobOrigin::User,
);
if handle.is_none() {
// A sentence of its own where it lands: this is a banner, not a clause.
*error = Some(busy::sentence(&busy::running(busy::IMPORT_OF_THAT_SOURCE)).into());
+1 -1
View File
@@ -1281,7 +1281,7 @@ impl EnsembleView {
self.active_jobs
.control(
JobKind::CatalogSync,
&JobSubject::Peer(peer),
&JobSubject::Peer { node: peer },
member.sync.is_pending(),
)
.busy()
+2 -2
View File
@@ -83,7 +83,7 @@ impl SettingsView {
let Some(handle) = self.job_runner.start(
JobKind::Eviction,
JobSubject::DocumentType(doc_type),
JobSubject::DocumentType { doc_type },
format!("Reclaim: {}", doc_type.display_name()),
JobOrigin::User,
) else {
@@ -133,7 +133,7 @@ impl SettingsView {
/// reclaim dedups against itself rather than colliding with — or being
/// masked by — a per-document eviction of something in that type.
fn reclaim_control(&self, doc_type: DocumentType) -> JobControl {
self.active_jobs.control(JobKind::Eviction, &JobSubject::DocumentType(doc_type), false)
self.active_jobs.control(JobKind::Eviction, &JobSubject::DocumentType { doc_type }, false)
}
/// A status line under a type's [`crate::components::sync_policy::SyncPolicyControl`]
+1 -1
View File
@@ -2274,7 +2274,7 @@ mod tests {
store.node_content_status_store.mark_present(
tx,
&node(1),
&ContentKey::Blob(blob_ref.blob_hash.clone()),
&ContentKey::Blob { hash: blob_ref.blob_hash.clone() },
)
})
.unwrap();
+1 -1
View File
@@ -955,7 +955,7 @@ mod tests {
store.node_content_status_store.mark_present(
tx,
&test_node(),
&ContentKey::Artifact(stub_id),
&ContentKey::Artifact { id: stub_id },
)
})
.unwrap();
+1 -1
View File
@@ -790,7 +790,7 @@ mod tests {
store.node_content_status_store.mark_present(
tx,
&node(1),
&ContentKey::Blob(blob_ref.blob_hash.clone()),
&ContentKey::Blob { hash: blob_ref.blob_hash.clone() },
)
})
.unwrap();
+1 -1
View File
@@ -470,7 +470,7 @@ pub fn start_stream_job(
job_runner.and_then(|runner| {
runner.start(
JobKind::ContentStream,
JobSubject::Document(doc_id.clone()),
JobSubject::Document { document_id: doc_id.clone() },
label,
JobOrigin::User,
)
+4 -3
View File
@@ -277,7 +277,8 @@ pub fn document_placement(
// as evidence rather than fact, which is strictly worse information about
// the one node that needs none.
let local = store.settings_store.local_node_pubkey().ok().flatten();
let keys: Vec<ContentKey> = refs.iter().map(|(h, _)| ContentKey::Blob(h.clone())).collect();
let keys: Vec<ContentKey> =
refs.iter().map(|(h, _)| ContentKey::Blob { hash: h.clone() }).collect();
let mut holders: HashMap<ContentKey, Vec<(NodePubkey, Timestamp)>> = HashMap::new();
for (key, node, at) in store
.node_content_status_store
@@ -290,7 +291,7 @@ pub fn document_placement(
}
for (hash, size) in &refs {
let Some(nodes) = holders.get(&ContentKey::Blob(hash.clone())) else {
let Some(nodes) = holders.get(&ContentKey::Blob { hash: hash.clone() }) else {
continue;
};
for (node, at) in nodes {
@@ -615,7 +616,7 @@ mod tests {
store.node_content_status_store.record_at(
tx,
&node,
&ContentKey::Blob(hash.clone()),
&ContentKey::Blob { hash: hash.clone() },
crate::stores::node_content_status::ContentPresence::Present,
at,
)
+8
View File
@@ -15,6 +15,7 @@ use std::str::FromStr;
use strum::{EnumString, EnumTryAs, IntoStaticStr, VariantArray};
use uuid::Uuid;
use crate::ffi::SyncError;
use crate::node::NodePubkey;
use crate::vault::VaultId;
@@ -642,6 +643,13 @@ impl fmt::Display for BlobHash {
}
}
// Crosses the FFI as its display form, `"blake3:<hex>"`.
uniffi::custom_type!(BlobHash, String, {
try_lift: |s| s.parse::<BlobHash>()
.map_err(|e| SyncError::InvalidInput { msg: format!("invalid blob hash: {e}") }.into()),
lower: |hash| hash.to_string(),
});
impl FromStr for BlobHash {
type Err = BlobHashError;
+9 -6
View File
@@ -170,7 +170,7 @@ pub fn artifact_availability_in_tx(
else {
return Ok(DocumentAvailability::Local);
};
let key = ContentKey::Artifact(stub.id);
let key = ContentKey::Artifact { id: stub.id };
let holders = node_content_store.keys_with_known_holder(tx, std::slice::from_ref(&key))?;
Ok(if holders.contains(&key) {
DocumentAvailability::Remote
@@ -223,7 +223,10 @@ pub fn document_availability_bulk(
// blob bytes we don't hold, and inline values we hold only a stub for.
let mut missing = missing_blobs(tx, artifact_store, blob_store, document_ids)?;
for (doc_id, stub_ids) in artifact_store.unfulfilled_stub_ids_for_documents(tx, document_ids)? {
missing.entry(doc_id).or_default().extend(stub_ids.into_iter().map(ContentKey::Artifact));
missing
.entry(doc_id)
.or_default()
.extend(stub_ids.into_iter().map(|id| ContentKey::Artifact { id }));
}
let all_missing: Vec<ContentKey> =
@@ -268,7 +271,7 @@ fn missing_blobs(
.iter()
.map(|(_, r)| r)
.filter(|r| !blob_store.exists(&r.blob_hash))
.map(|r| ContentKey::Blob(r.blob_hash.clone()))
.map(|r| ContentKey::Blob { hash: r.blob_hash.clone() })
.collect();
(doc_id, keys)
})
@@ -325,7 +328,7 @@ mod tests {
#[rstest]
fn missing_blob_with_a_known_holder_is_remote(store: TestStore) {
let (id, hash) = doc_with_missing_blob(&store);
record_holder(&store, ContentKey::Blob(hash.clone()));
record_holder(&store, ContentKey::Blob { hash: hash.clone() });
assert_eq!(availability_of(&store, &id), DocumentAvailability::Remote);
}
@@ -356,7 +359,7 @@ mod tests {
test_node(),
);
if with_holder {
record_holder(&store, ContentKey::Artifact(stub_id));
record_holder(&store, ContentKey::Artifact { id: stub_id });
}
assert_eq!(availability_of(&store, &id), expected);
}
@@ -366,7 +369,7 @@ mod tests {
#[rstest]
fn one_unsourceable_blob_outweighs_a_fetchable_sibling(store: TestStore) {
let (id, hash) = doc_with_missing_blob(&store);
record_holder(&store, ContentKey::Blob(hash.clone()));
record_holder(&store, ContentKey::Blob { hash: hash.clone() });
upsert_artifact(
&store.artifact_store,
&Artifact::new(
+9
View File
@@ -26,6 +26,15 @@ uniffi::custom_type!(DocumentID, String, {
lower: |id| id.to_string(),
});
// A bare `Uuid` — an artifact's id — crosses as its hyphenated string, the
// same form `DocumentID` takes.
uniffi::custom_type!(Uuid, String, {
remote,
try_lift: |s| Uuid::parse_str(&s)
.map_err(|e| SyncError::InvalidInput { msg: format!("invalid id: {e}") }.into()),
lower: |id| id.to_string(),
});
impl DocumentID {
#[allow(clippy::new_without_default)]
pub fn new() -> Self {
+3 -1
View File
@@ -507,7 +507,9 @@ mod tests {
if with_holder {
let content = &store.node_content_status_store;
content
.batch(|tx| content.mark_present(tx, &node(1), &ContentKey::Artifact(stub_id)))
.batch(|tx| {
content.mark_present(tx, &node(1), &ContentKey::Artifact { id: stub_id })
})
.unwrap();
}
+11 -9
View File
@@ -324,7 +324,7 @@ fn recorded_evidence(
for hash_bytes in hashes {
let hash = BlobHash::Blake3(*hash_bytes);
if let Ok(nodes) =
store.node_content_status_store.nodes_holding(tx, &ContentKey::Blob(hash))
store.node_content_status_store.nodes_holding(tx, &ContentKey::Blob { hash })
&& !nodes.is_empty()
{
recorded.insert(*hash_bytes, nodes);
@@ -488,7 +488,7 @@ pub fn record_local_release(
if !deleted.is_empty() {
store.content_release_store.batch(|tx| {
for hash in deleted {
store.content_release_store.record(tx, &ContentKey::Blob(hash.clone()))?;
store.content_release_store.record(tx, &ContentKey::Blob { hash: hash.clone() })?;
}
Ok::<_, crate::stores::content_release::ContentReleaseError>(())
})?;
@@ -599,7 +599,7 @@ mod tests {
store.node_content_status_store.mark_present(
tx,
&node(0x01),
&ContentKey::Blob(BlobHash::Blake3(known)),
&ContentKey::Blob { hash: BlobHash::Blake3(known) },
)
})
.unwrap();
@@ -679,11 +679,13 @@ mod tests {
releases
.batch(|tx| {
assert!(
releases.released_at(tx, &ContentKey::Blob(deleted.clone()))?.is_some(),
releases
.released_at(tx, &ContentKey::Blob { hash: deleted.clone() })?
.is_some(),
"an evicted blob must be announceable to peers"
);
assert!(
releases.released_at(tx, &ContentKey::Blob(kept.clone()))?.is_none(),
releases.released_at(tx, &ContentKey::Blob { hash: kept.clone() })?.is_none(),
"a blob we still hold must never be announced as released"
);
Ok::<_, crate::stores::content_release::ContentReleaseError>(())
@@ -705,7 +707,7 @@ mod tests {
let releases = &store.content_release_store;
assert!(
releases
.batch(|tx| releases.released_at(tx, &ContentKey::Blob(hash)))
.batch(|tx| releases.released_at(tx, &ContentKey::Blob { hash }))
.unwrap()
.is_none(),
"a skipped blob is still here; announcing its release would be a lie"
@@ -725,10 +727,10 @@ mod tests {
let evidence = &store.node_content_status_store;
evidence
.batch(|tx| {
evidence.mark_present(tx, &node(0x01), &ContentKey::Blob(hash.clone()))?;
evidence.mark_present(tx, &node(0x02), &ContentKey::Blob(hash.clone()))?;
evidence.mark_present(tx, &node(0x01), &ContentKey::Blob { hash: hash.clone() })?;
evidence.mark_present(tx, &node(0x02), &ContentKey::Blob { hash: hash.clone() })?;
// Node 2 later told us it dropped its copy.
evidence.mark_absent(tx, &node(0x02), &ContentKey::Blob(hash.clone()))
evidence.mark_absent(tx, &node(0x02), &ContentKey::Blob { hash: hash.clone() })
})
.unwrap();
+1 -1
View File
@@ -408,7 +408,7 @@ impl DocumentsFfi {
Ok::<_, crate::stores::node_content_status::NodeContentStatusError>(
refs.into_iter()
.map(|(kind, r)| {
let key = ContentKey::Blob(r.blob_hash.clone());
let key = ContentKey::Blob { hash: r.blob_hash.clone() };
let released_at =
store.content_release_store.released_at(tx, &key).ok().flatten();
let node_statuses = node_records
+6 -11
View File
@@ -15,9 +15,7 @@ use crate::stores::node_content_status::ContentKey;
use super::error::SyncError;
use super::free_fns::{
DocumentAssetRef, DocumentPushResult, DocumentSyncResult, PurgeResult, SyncDocumentsResult,
};
use super::free_fns::{DocumentPushResult, DocumentSyncResult, PurgeResult, SyncDocumentsResult};
use super::node::SyncNodeCore;
/// A revocation this node just minted or imported, for the screen that asked
@@ -534,7 +532,7 @@ impl EnsembleFfi {
let res = run_as_job_with_progress(
&self.core.job_runner,
JobKind::ContentFetch,
JobSubject::Document(document_id.clone()),
JobSubject::Document { document_id: document_id.clone() },
format!("Pull {document_id}"),
|reporter| {
rt_handle
@@ -559,9 +557,7 @@ impl EnsembleFfi {
///
/// The narrow counterpart to [`Self::sync_document_assets`], which stays
/// the bulk action: this fetches exactly the named artifact and does not
/// recurse into child documents. `asset` is a [`DocumentAssetRef`], whose
/// two variants the frontends already have in hand from the rows they
/// render.
/// recurse into child documents.
///
/// `artifacts_requested` is 0 when the artifact is already present
/// locally, so a frontend can distinguish "nothing to do" from "fetched"
@@ -570,9 +566,8 @@ impl EnsembleFfi {
pub fn sync_document_asset(
&self,
document_id: DocumentID,
asset: DocumentAssetRef,
asset: ContentKey,
) -> Result<DocumentSyncResult, SyncError> {
let asset: ContentKey = asset.try_into()?;
let peer_ids = self.ensemble_peer_ids()?;
let (endpoint, rt_handle, store) = self.core.require_p2p()?;
@@ -583,7 +578,7 @@ impl EnsembleFfi {
let res = run_as_job_with_progress(
&self.core.job_runner,
JobKind::ContentFetch,
JobSubject::DocumentAsset(document_id.clone(), asset.clone()),
JobSubject::DocumentAsset { document_id: document_id.clone(), asset: asset.clone() },
format!("Pull {asset} of {document_id}"),
|reporter| {
rt_handle
@@ -622,7 +617,7 @@ impl EnsembleFfi {
let res = run_as_job_with_progress(
&self.core.job_runner,
JobKind::Push,
JobSubject::DocumentToPeer(document_id.clone(), peer),
JobSubject::DocumentToPeer { document_id: document_id.clone(), node: peer },
format!("Push {document_id} to {}", peer.to_hex()),
|reporter| {
rt_handle
+4 -1
View File
@@ -662,8 +662,9 @@ mod tests {
use uniffi::{Lift, Lower};
use crate::UniFfiTag;
use crate::documents::{DocumentID, Timestamp};
use crate::documents::{BlobHash, DocumentID, Timestamp};
use crate::node::NodePubkey;
use uuid::Uuid;
use super::*;
@@ -677,5 +678,7 @@ mod tests {
assert!(invalid(<NodePubkey as Lift<UniFfiTag>>::try_lift(wire()).unwrap_err()));
assert!(invalid(<DocumentID as Lift<UniFfiTag>>::try_lift(wire()).unwrap_err()));
assert!(invalid(<Timestamp as Lift<UniFfiTag>>::try_lift(wire()).unwrap_err()));
assert!(invalid(<BlobHash as Lift<UniFfiTag>>::try_lift(wire()).unwrap_err()));
assert!(invalid(<Uuid as Lift<UniFfiTag>>::try_lift(wire()).unwrap_err()));
}
}
+2 -2
View File
@@ -63,7 +63,7 @@ impl EvictionFfi {
let outcome = run_as_job(
&self.core.job_runner,
JobKind::Eviction,
JobSubject::Document(document_id.clone()),
JobSubject::Document { document_id: document_id.clone() },
format!("Delete locally: {document_id}"),
|| -> Result<eviction::EvictionOutcome, SyncError> {
let plan = rt_handle.block_on(eviction::evaluate_subtree_eviction(
@@ -108,7 +108,7 @@ impl EvictionFfi {
let outcome = run_as_job(
&self.core.job_runner,
JobKind::Eviction,
JobSubject::DocumentType(doc_type),
JobSubject::DocumentType { doc_type },
format!("Reclaim: {}", doc_type.display_name()),
|| -> Result<eviction::EvictionOutcome, SyncError> {
let plan = rt_handle
+10 -181
View File
@@ -6,15 +6,13 @@ use crate::copy::ContentRowCopy;
use crate::copy::content_fetch::{FetchCopy, FetchDisplay};
use crate::copy::peer_health::PeerHealthCopy;
use crate::copy::{self};
use crate::documents::{BlobHash, DocumentAvailability, DocumentID, DocumentType, TextFormat};
use crate::documents::{DocumentAvailability, DocumentID, DocumentType, TextFormat};
use crate::ensemble::screen::{EnsembleStatus, Roster};
use crate::ensemble::secret::{EnsembleSecret, Secret};
use crate::jobs::{JobKind, JobSubject};
use crate::log::LogVerbosity;
use crate::node::NodePubkey;
use crate::p2p::node_admin;
use crate::scheduler::SchedulerPassResult;
use crate::stores::node_content_status::ContentKey;
use crate::stores::peer_health::PeerHealthIssue;
use crate::sync_prefs::{DocumentSyncOverride, SyncPolicy};
use strum::VariantArray;
@@ -407,59 +405,6 @@ pub fn job_kind_label(kind: JobKind) -> String {
kind.display_name().to_string()
}
/// A job's subject, in the shapes a frontend already has in hand — the FFI
/// spelling of [`JobSubject`], and the argument to [`job_match_key`].
///
/// Only the variants a widget can bind to are here — `JobSubject::Path` is a
/// folder import's tree URI, and nothing else in [`JobSubject`] is missing.
#[derive(uniffi::Enum)]
pub enum JobSubjectRef {
/// One peer node, by its hex node id — a `CatalogSync`.
Peer { node_id_hex: NodePubkey },
/// One document as a whole — a `ContentFetch` of everything it and its
/// descendants are missing.
Document { document_id: DocumentID },
/// One document pushed to one peer.
DocumentToPeer { document_id: DocumentID, node_id_hex: NodePubkey },
/// One artifact of one document, pulled on its own.
DocumentAsset { document_id: DocumentID, asset: DocumentAssetRef },
/// Every document of one type — a type-scoped `Eviction` (the Sync policy
/// tab's "Reclaim…") or a `LibraryBuild`.
///
/// The enum itself rather than its snake_case name: the name is Rust's
/// spelling, and a frontend that had to produce one would be a mirror of
/// `DocumentType::type_name` — exactly what `job_match_key` exists to
/// avoid. Kotlin already holds `DocumentType` values.
DocumentType { doc_type: DocumentType },
/// A filesystem path or content URI — a folder import's tree URI.
Path { path: String },
/// Work about the node as a whole rather than any one thing in it: a
/// `BackgroundSync` pass. Its widget is the Settings screen's "Run now",
/// which speaks for whichever pass is running — the periodic tick's, a
/// peer's `TriggerSync`, or its own.
None,
}
impl TryFrom<JobSubjectRef> for JobSubject {
type Error = SyncError;
fn try_from(value: JobSubjectRef) -> Result<Self, SyncError> {
Ok(match value {
JobSubjectRef::Peer { node_id_hex } => JobSubject::Peer(node_id_hex),
JobSubjectRef::Document { document_id } => JobSubject::Document(document_id),
JobSubjectRef::DocumentToPeer { document_id, node_id_hex } => {
JobSubject::DocumentToPeer(document_id, node_id_hex)
}
JobSubjectRef::DocumentAsset { document_id, asset } => {
JobSubject::DocumentAsset(document_id, asset.try_into()?)
}
JobSubjectRef::DocumentType { doc_type } => JobSubject::DocumentType(doc_type),
JobSubjectRef::Path { path } => JobSubject::Path(path),
JobSubjectRef::None => JobSubject::None,
})
}
}
/// The key a widget compares against [`JobItem::match_key`] to find "its" job
/// in the active list — `(kind, subject)`, the same pair
/// [`JobRunner::start`](crate::jobs::runner::JobRunner::start) dedups on.
@@ -471,9 +416,13 @@ impl TryFrom<JobSubjectRef> for JobSubject {
///
/// The subject alone is not enough to bind to: an `Eviction` and a
/// `ContentFetch` of the same document share a subject and are different work.
///
/// Nothing here fails; the `Result` is for the argument. A malformed id or
/// hash in `subject` fails as uniffi lifts it, and only a function returning
/// `SyncError` hands that to Kotlin as a typed `InvalidInput`.
#[uniffi::export]
pub fn job_match_key(kind: JobKind, subject: JobSubjectRef) -> Result<String, SyncError> {
Ok(crate::jobs::match_key(kind, &JobSubject::try_from(subject)?))
pub fn job_match_key(kind: JobKind, subject: JobSubject) -> Result<String, SyncError> {
Ok(crate::jobs::match_key(kind, &subject))
}
/// What a widget running a job says about it — [`copy::job_activity`].
@@ -1202,43 +1151,6 @@ impl From<crate::p2p::asset_sync::DocAssetSyncResult> for DocumentSyncResult {
}
}
/// One artifact of a document, named so it can be synced on its own — the
/// argument to [`EnsembleFfi::sync_document_asset`], and the FFI spelling of
/// [`ContentKey`].
///
/// Two variants because artifacts have two storage tiers and each is addressed
/// by what identifies it: blob-backed content is content-addressed, so the
/// hash is the identity and evidence about it is shared by every artifact
/// pointing at the same bytes; an inline value has no identity apart from the
/// artifact row holding it. Frontends already have both — a blob row shows
/// [`DocumentBlobRefItem::blob_hash`], an inline row
/// [`DocumentArtifactItem::id`].
#[derive(uniffi::Enum)]
pub enum DocumentAssetRef {
/// Blob ciphertext, by its full hash string (`"blake3:<hex>"`), exactly as
/// [`DocumentBlobRefItem::blob_hash`] reports it.
Blob { hash: String },
/// One inline artifact's value, by the artifact's stable id (a UUID).
Artifact { id: String },
}
impl TryFrom<DocumentAssetRef> for ContentKey {
type Error = SyncError;
fn try_from(value: DocumentAssetRef) -> Result<Self, SyncError> {
match value {
DocumentAssetRef::Blob { hash } => hash
.parse::<BlobHash>()
.map(ContentKey::Blob)
.map_err(|e| SyncError::InvalidInput { msg: format!("invalid blob hash: {e}") }),
DocumentAssetRef::Artifact { id } => id
.parse::<uuid::Uuid>()
.map(ContentKey::Artifact)
.map_err(|e| SyncError::InvalidInput { msg: format!("invalid artifact id: {e}") }),
}
}
}
/// Summary returned by [`EnsembleFfi::push_document`]. Mirrors
/// [`crate::p2p::asset_sync::DocPushResult`], the write-side counterpart to
/// [`DocumentSyncResult`].
@@ -1262,101 +1174,18 @@ pub struct DocumentPushResult {
#[cfg(test)]
mod tests {
use super::*;
use crate::documents::Timestamp;
use crate::jobs::{Job, JobOrigin, JobState};
use crate::test_support::node;
use rstest::rstest;
use std::collections::HashMap;
/// A widget builds its key from the ids it holds; a job row's key is built
/// from the parsed subject. The two are the same string or the widget
/// silently never finds its own job — so this is the drift guard for every
/// subject shape a widget can bind to.
#[rstest]
#[case(
JobSubjectRef::Document { document_id: doc() },
JobKind::ContentFetch,
JobSubject::Document(doc()),
)]
#[case(
JobSubjectRef::DocumentToPeer { document_id: doc(), node_id_hex: node(7) },
JobKind::Push,
JobSubject::DocumentToPeer(doc(), node(7)),
)]
#[case(
JobSubjectRef::DocumentAsset {
document_id: doc(),
asset: DocumentAssetRef::Artifact { id: ARTIFACT.to_string() },
},
JobKind::ContentFetch,
JobSubject::DocumentAsset(doc(), ContentKey::Artifact(artifact())),
)]
#[case(
JobSubjectRef::Peer { node_id_hex: node(7) },
JobKind::CatalogSync,
JobSubject::Peer(node(7)),
)]
#[case(
JobSubjectRef::Path { path: "content://tree/Music".to_string() },
JobKind::Import,
JobSubject::Path("content://tree/Music".to_string()),
)]
#[case(
JobSubjectRef::DocumentType { doc_type: DocumentType::MusicTrack },
JobKind::Eviction,
JobSubject::DocumentType(DocumentType::MusicTrack),
)]
#[case(JobSubjectRef::None, JobKind::BackgroundSync, JobSubject::None)]
fn a_widgets_key_is_the_key_on_its_job_row(
#[case] widget_subject: JobSubjectRef,
#[case] kind: JobKind,
#[case] job_subject: JobSubject,
) {
let from_widget = job_match_key(kind, widget_subject).unwrap();
let row = super::super::types::job_to_item(
job(kind, job_subject),
&HashMap::new(),
&HashMap::new(),
);
assert_eq!(from_widget, row.match_key);
}
/// Same subject, different work: an eviction of a document must not look
/// to the "Sync to this node" button like the pull it started.
#[rstest]
#[test]
fn kind_is_part_of_the_identity() {
let subject = || JobSubjectRef::Document { document_id: doc() };
let document_id = DocumentID::new();
let subject = || JobSubject::Document { document_id: document_id.clone() };
assert_ne!(
job_match_key(JobKind::ContentFetch, subject()).unwrap(),
job_match_key(JobKind::Eviction, subject()).unwrap(),
);
}
const DOC: &str = "01998f3e-0000-7000-8000-000000000001";
const ARTIFACT: &str = "01998f3e-0000-7000-8000-000000000002";
fn doc() -> DocumentID {
DOC.parse().expect("valid document id")
}
fn artifact() -> uuid::Uuid {
ARTIFACT.parse().expect("valid uuid")
}
fn job(kind: JobKind, subject: JobSubject) -> Job {
Job {
id: uuid::Uuid::nil(),
kind,
subject,
label: "label".to_string(),
origin: JobOrigin::User,
state: JobState::Running,
progress: None,
created_at: Timestamp::now(),
started_at: None,
finished_at: None,
}
}
}
/// What Android draws for one entry in a peer's sync-health list —
+7 -2
View File
@@ -81,7 +81,12 @@ impl ImportFfi {
let handle = self
.core
.job_runner
.start(JobKind::Import, JobSubject::Path(source), label.clone(), JobOrigin::User)
.start(
JobKind::Import,
JobSubject::Path { path: source },
label.clone(),
JobOrigin::User,
)
.ok_or_else(|| SyncError::Busy {
msg: busy::sentence(&busy::running(&busy::import_of(&format!("\"{label}\"")))),
})?;
@@ -115,7 +120,7 @@ impl ImportFfi {
run_as_job_with_progress(
&self.core.job_runner,
JobKind::Import,
JobSubject::Path(url.clone()),
JobSubject::Path { path: url.clone() },
url.clone(),
|reporter| {
let progress = ImportProgress::files(Some(reporter), Some(ImportTotals::file(0)));
+1 -1
View File
@@ -106,7 +106,7 @@ impl MusicFfi {
.job_runner
.start(
JobKind::LibraryBuild,
JobSubject::DocumentType(DocumentType::MusicTrack),
JobSubject::DocumentType { doc_type: DocumentType::MusicTrack },
"Build music library",
JobOrigin::User,
)
+1 -1
View File
@@ -180,7 +180,7 @@ impl PodcastFfi {
run_as_job_with_progress(
&self.core.job_runner,
JobKind::Import,
JobSubject::Document(document_id),
JobSubject::Document { document_id },
copy::episode_download::job_label(&episode.title),
|reporter| {
// One file, of a size nobody knows until the response arrives
+5 -3
View File
@@ -92,8 +92,9 @@ pub struct DocumentPlacementRow {
#[derive(uniffi::Record)]
pub struct DocumentArtifactItem {
/// Stable artifact id (UUID). Addresses this artifact for a per-row sync —
/// see [`DocumentAssetRef::Artifact`] — and is the only thing that tells
/// two members of a collection kind apart, since `kind` doesn't.
/// see [`ContentKey::Artifact`](crate::stores::node_content_status::ContentKey::Artifact)
/// — and is the only thing that tells two members of a collection kind
/// apart, since `kind` doesn't.
pub id: String,
/// Artifact kind name (e.g. `"note_body"`, `"listen_progress"`).
pub kind: String,
@@ -113,7 +114,8 @@ pub struct DocumentBlobRefItem {
/// Blob-backed artifact kind name (e.g. `"primary_content"`, `"thumbnail"`).
pub kind: String,
/// Full hash string (e.g. `"blake3:abcdef…"`). Never truncated. Also
/// addresses this blob for a per-row sync — see [`DocumentAssetRef::Blob`].
/// addresses this blob for a per-row sync — see
/// [`ContentKey::Blob`](crate::stores::node_content_status::ContentKey::Blob).
pub blob_hash: String,
pub size_bytes: u64,
/// Whether this node holds the ciphertext, as opposed to just the pointer
+60 -40
View File
@@ -91,26 +91,35 @@ impl JobKind {
/// [`Job::label`] for that. Two jobs of the same [`JobKind`] with an equal
/// `JobSubject` are considered "the same work"; see
/// [`runner::JobRunner::start`].
#[derive(Debug, Clone, PartialEq, Eq)]
///
/// Also what a frontend widget names to find "its" job in the active list —
/// see [`job_match_key`](crate::ffi::job_match_key).
#[derive(Debug, Clone, PartialEq, Eq, uniffi::Enum)]
pub enum JobSubject {
Peer(NodePubkey),
Document(DocumentID),
/// One peer node — a `CatalogSync`.
Peer { node: NodePubkey },
/// One document as a whole — e.g. a `ContentFetch` of everything it and
/// its descendants are missing.
Document { document_id: DocumentID },
/// A document pushed to one specific peer — distinct from `Document`
/// alone so pushing the same document to two different peers (each its
/// own row on the detail screen) doesn't dedup against each other.
DocumentToPeer(DocumentID, NodePubkey),
DocumentToPeer { document_id: DocumentID, node: NodePubkey },
/// One artifact of a document pulled on its own — the per-row "Sync"
/// affordance in the detail screen's Artifacts section. Distinct from
/// `Document` for the same reason `DocumentToPeer` is: each artifact row
/// starts its own fetch, so syncing two of them (or one of them while a
/// whole-document pull is running) must not dedup against each other.
DocumentAsset(DocumentID, ContentKey),
DocumentAsset { document_id: DocumentID, asset: ContentKey },
/// Every document of one type — the Sync policy tab's type-scoped
/// "Reclaim…" eviction, distinct from `Document` so it dedups against
/// itself rather than colliding with (or being masked by) a per-document
/// eviction job.
DocumentType(DocumentType),
Path(String),
/// eviction job. Also a `LibraryBuild`.
DocumentType { doc_type: DocumentType },
/// A filesystem path or content URI — a folder import's tree URI.
Path { path: String },
/// Work about the node as a whole rather than any one thing in it: a
/// `BackgroundSync` pass, whichever trigger started it.
None,
}
@@ -118,12 +127,12 @@ impl JobSubject {
/// The document this work is about, when it is about one.
pub fn document(&self) -> Option<&DocumentID> {
match self {
JobSubject::Document(id)
| JobSubject::DocumentToPeer(id, _)
| JobSubject::DocumentAsset(id, _) => Some(id),
JobSubject::Peer(_)
| JobSubject::DocumentType(_)
| JobSubject::Path(_)
JobSubject::Document { document_id: id }
| JobSubject::DocumentToPeer { document_id: id, .. }
| JobSubject::DocumentAsset { document_id: id, .. } => Some(id),
JobSubject::Peer { .. }
| JobSubject::DocumentType { .. }
| JobSubject::Path { .. }
| JobSubject::None => None,
}
}
@@ -131,11 +140,13 @@ impl JobSubject {
/// The peer this work is with, when it is with one.
pub fn peer(&self) -> Option<NodePubkey> {
match self {
JobSubject::Peer(pk) | JobSubject::DocumentToPeer(_, pk) => Some(*pk),
JobSubject::Document(_)
| JobSubject::DocumentAsset(..)
| JobSubject::DocumentType(_)
| JobSubject::Path(_)
JobSubject::Peer { node: pk } | JobSubject::DocumentToPeer { node: pk, .. } => {
Some(*pk)
}
JobSubject::Document { .. }
| JobSubject::DocumentAsset { .. }
| JobSubject::DocumentType { .. }
| JobSubject::Path { .. }
| JobSubject::None => None,
}
}
@@ -144,12 +155,18 @@ impl JobSubject {
/// checks that don't need the parsed enum back.
pub fn to_key(&self) -> String {
match self {
JobSubject::Peer(pk) => format!("peer:{}", pk.to_hex()),
JobSubject::Document(id) => format!("document:{id}"),
JobSubject::DocumentToPeer(id, pk) => format!("document_to_peer:{id}:{}", pk.to_hex()),
JobSubject::DocumentAsset(id, asset) => format!("document_asset:{id}:{asset}"),
JobSubject::DocumentType(doc_type) => format!("document_type:{}", doc_type.type_name()),
JobSubject::Path(p) => format!("path:{p}"),
JobSubject::Peer { node: pk } => format!("peer:{}", pk.to_hex()),
JobSubject::Document { document_id: id } => format!("document:{id}"),
JobSubject::DocumentToPeer { document_id: id, node: pk } => {
format!("document_to_peer:{id}:{}", pk.to_hex())
}
JobSubject::DocumentAsset { document_id: id, asset } => {
format!("document_asset:{id}:{asset}")
}
JobSubject::DocumentType { doc_type } => {
format!("document_type:{}", doc_type.type_name())
}
JobSubject::Path { path: p } => format!("path:{p}"),
JobSubject::None => "none".to_string(),
}
}
@@ -161,13 +178,13 @@ impl JobSubject {
}
if let Some(hex) = s.strip_prefix("peer:") {
let pk = NodePubkey::from_str(hex).map_err(|_| bad())?;
return Ok(JobSubject::Peer(pk));
return Ok(JobSubject::Peer { node: pk });
}
if let Some(rest) = s.strip_prefix("document_to_peer:") {
let (id, hex) = rest.split_once(':').ok_or_else(bad)?;
let id = DocumentID::from_str(id).map_err(|_| bad())?;
let pk = NodePubkey::from_str(hex).map_err(|_| bad())?;
return Ok(JobSubject::DocumentToPeer(id, pk));
return Ok(JobSubject::DocumentToPeer { document_id: id, node: pk });
}
if let Some(rest) = s.strip_prefix("document_asset:") {
// The document id is a uuid and so contains no `:`; everything
@@ -176,18 +193,18 @@ impl JobSubject {
let (id, asset) = rest.split_once(':').ok_or_else(bad)?;
let id = DocumentID::from_str(id).map_err(|_| bad())?;
let asset = ContentKey::from_str(asset).map_err(|_| bad())?;
return Ok(JobSubject::DocumentAsset(id, asset));
return Ok(JobSubject::DocumentAsset { document_id: id, asset });
}
if let Some(id) = s.strip_prefix("document:") {
let id = DocumentID::from_str(id).map_err(|_| bad())?;
return Ok(JobSubject::Document(id));
return Ok(JobSubject::Document { document_id: id });
}
if let Some(type_name) = s.strip_prefix("document_type:") {
let doc_type = DocumentType::from_str(type_name).map_err(|_| bad())?;
return Ok(JobSubject::DocumentType(doc_type));
return Ok(JobSubject::DocumentType { doc_type });
}
if let Some(path) = s.strip_prefix("path:") {
return Ok(JobSubject::Path(path.to_string()));
return Ok(JobSubject::Path { path: path.to_string() });
}
Err(bad())
}
@@ -564,10 +581,13 @@ mod tests {
#[rstest]
fn run_as_job_records_done_on_success(store: TestStore) {
let runner = JobRunner::new(store.job_store.clone());
let result =
run_as_job(&runner, JobKind::Import, JobSubject::Path("a.pdf".into()), "a.pdf", || {
Ok::<_, TestError>(42)
});
let result = run_as_job(
&runner,
JobKind::Import,
JobSubject::Path { path: "a.pdf".into() },
"a.pdf",
|| Ok::<_, TestError>(42),
);
assert_eq!(result, Ok(42));
let history = store.job_store.batch(|tx| store.job_store.list_history(tx, 10)).unwrap();
@@ -584,7 +604,7 @@ mod tests {
let result = run_as_job_with_progress(
&runner,
JobKind::Import,
JobSubject::Path("photos".into()),
JobSubject::Path { path: "photos".into() },
"photos",
|reporter| {
for done in 1..=2 {
@@ -615,7 +635,7 @@ mod tests {
#[rstest]
fn a_second_identical_job_is_refused_while_the_first_runs(store: TestStore) {
let runner = JobRunner::new(store.job_store.clone());
let subject = JobSubject::Path("photos".into());
let subject = JobSubject::Path { path: "photos".into() };
let outer: Result<(), TestError> =
run_as_job(&runner, JobKind::Import, subject.clone(), "photos", || {
@@ -643,7 +663,7 @@ mod tests {
let result: Result<(), TestError> = run_as_job_with_progress(
&runner,
JobKind::Import,
JobSubject::Path("photos".into()),
JobSubject::Path { path: "photos".into() },
"photos",
|reporter| {
for _ in 0..10 {
@@ -652,7 +672,7 @@ mod tests {
}
files_seen += 1;
let id = runner
.find_active(JobKind::Import, &JobSubject::Path("photos".into()))
.find_active(JobKind::Import, &JobSubject::Path { path: "photos".into() })
.expect("a wrapped job is visible to the runner while it runs");
assert!(runner.cancel(id), "cancel found a live token");
}
@@ -725,7 +745,7 @@ mod tests {
run_as_job_with_progress(
&runner,
JobKind::Import,
JobSubject::Path("photos".into()),
JobSubject::Path { path: "photos".into() },
"photos",
|reporter| {
reporter.report(JobProgress {
+8 -3
View File
@@ -365,7 +365,7 @@ mod tests {
use rstest::rstest;
fn peer(b: u8) -> JobSubject {
JobSubject::Peer(node(b))
JobSubject::Peer { node: node(b) }
}
#[rstest]
@@ -450,7 +450,12 @@ mod tests {
fn dropping_handle_without_finishing_marks_failed(store: TestStore) {
let runner = JobRunner::new(store.job_store.clone());
let handle = runner
.start(JobKind::Import, JobSubject::Path("/tmp/x".into()), "import", JobOrigin::User)
.start(
JobKind::Import,
JobSubject::Path { path: "/tmp/x".into() },
"import",
JobOrigin::User,
)
.unwrap();
let id = handle.id();
drop(handle);
@@ -464,7 +469,7 @@ mod tests {
// And the dedup entry is cleared too, since drop calls finish_with_state.
let again = runner.start(
JobKind::Import,
JobSubject::Path("/tmp/x".into()),
JobSubject::Path { path: "/tmp/x".into() },
"import again",
JobOrigin::User,
);
+4 -4
View File
@@ -284,7 +284,7 @@ impl ArtifactSyncHandler {
self.store.node_content_status_store.mark_present(
tx,
&pusher,
&ContentKey::Artifact(artifact.id),
&ContentKey::Artifact { id: artifact.id },
)?;
applied += 1;
}
@@ -302,7 +302,7 @@ impl ArtifactSyncHandler {
/// Record whether `peer` holds artifact `id`'s value. Failures are logged,
/// never propagated: bookkeeping must not fail a transfer that worked.
fn record_peer_presence(store: &Store, peer: NodePubkey, id: Uuid, held: bool) {
let key = ContentKey::Artifact(id);
let key = ContentKey::Artifact { id };
let evidence = &store.node_content_status_store;
let result = evidence.batch(|tx| {
if held {
@@ -469,7 +469,7 @@ mod tests {
record_peer_presence(&store, node(1), id, held);
let evidence = &store.node_content_status_store;
let recorded = evidence
.batch(|tx| evidence.presence_of(tx, &node(1), &ContentKey::Artifact(id)))
.batch(|tx| evidence.presence_of(tx, &node(1), &ContentKey::Artifact { id }))
.unwrap();
assert_eq!(recorded.map(|(presence, _)| presence), Some(expected));
}
@@ -481,7 +481,7 @@ mod tests {
record_peer_presence(&store, node(1), id, false);
let evidence = &store.node_content_status_store;
let holders =
evidence.batch(|tx| evidence.nodes_holding(tx, &ContentKey::Artifact(id))).unwrap();
evidence.batch(|tx| evidence.nodes_holding(tx, &ContentKey::Artifact { id })).unwrap();
assert!(holders.is_empty());
}
}
+12 -9
View File
@@ -624,7 +624,7 @@ impl From<AssetPlan> for Missing {
fn plan_asset(store: &Store, doc_id: &DocumentID, asset: &ContentKey) -> Result<AssetPlan> {
let unknown = || AssetSyncError::UnknownAsset { doc_id: doc_id.clone(), asset: asset.clone() };
match asset {
ContentKey::Blob(hash) => {
ContentKey::Blob { hash } => {
let refs =
store.artifact_store.batch(|tx| store.artifact_store.list_blob_refs(tx, doc_id))?;
if !refs.iter().any(|(_, r)| &r.blob_hash == hash) {
@@ -636,7 +636,7 @@ fn plan_asset(store: &Store, doc_id: &DocumentID, asset: &ContentKey) -> Result<
let BlobHash::Blake3(bytes) = hash;
Ok(AssetPlan::Blob(*bytes))
}
ContentKey::Artifact(id) => {
ContentKey::Artifact { id } => {
let (stub, local) = store.artifact_store.batch(|tx| {
Ok::<_, ArtifactStoreError>((
store
@@ -922,14 +922,14 @@ mod tests {
#[rstest]
fn a_locally_held_artifact_needs_no_fetch(store: TestStore) {
let (doc_id, artifact_id) = doc_with_local_note(&store);
let plan = plan_asset(&store, &doc_id, &ContentKey::Artifact(artifact_id)).unwrap();
let plan = plan_asset(&store, &doc_id, &ContentKey::Artifact { id: artifact_id }).unwrap();
assert!(matches!(plan, AssetPlan::Present));
}
#[rstest]
fn a_stubbed_artifact_resolves_to_the_stub_to_fetch(store: TestStore) {
let (doc_id, artifact_id) = doc_with_stubbed_note(&store);
let plan = plan_asset(&store, &doc_id, &ContentKey::Artifact(artifact_id)).unwrap();
let plan = plan_asset(&store, &doc_id, &ContentKey::Artifact { id: artifact_id }).unwrap();
let AssetPlan::Artifact(stub) = plan else { panic!("expected a stub to fetch: {plan:?}") };
assert_eq!(stub.id, artifact_id);
assert_eq!(stub.document_id, doc_id);
@@ -941,15 +941,17 @@ mod tests {
let other = Document::new(DocumentType::Note, None, test_node());
let other_id = add_document(&store.doc_store, other);
let err = plan_asset(&store, &other_id, &ContentKey::Artifact(artifact_id)).unwrap_err();
let err =
plan_asset(&store, &other_id, &ContentKey::Artifact { id: artifact_id }).unwrap_err();
assert!(matches!(err, AssetSyncError::UnknownAsset { .. }), "{err:?}");
}
#[rstest]
fn an_unknown_artifact_id_is_rejected_rather_than_fetched(store: TestStore) {
let (doc_id, _) = doc_with_local_note(&store);
let err = plan_asset(&store, &doc_id, &ContentKey::Artifact(Uuid::from_bytes([3; 16])))
.unwrap_err();
let err =
plan_asset(&store, &doc_id, &ContentKey::Artifact { id: Uuid::from_bytes([3; 16]) })
.unwrap_err();
assert!(matches!(err, AssetSyncError::UnknownAsset { .. }), "{err:?}");
}
@@ -987,7 +989,7 @@ mod tests {
) {
let store = initialized_store;
let (doc_id, hash) = doc_with_blob(&store, stored);
let plan = plan_asset(&store, &doc_id, &ContentKey::Blob(hash)).unwrap();
let plan = plan_asset(&store, &doc_id, &ContentKey::Blob { hash }).unwrap();
assert_eq!(matches!(plan, AssetPlan::Present), stored, "{plan:?}");
}
@@ -995,7 +997,8 @@ mod tests {
fn a_hash_this_document_does_not_reference_is_rejected(store: TestStore) {
let (doc_id, _) = doc_with_blob(&store, false);
let err =
plan_asset(&store, &doc_id, &ContentKey::Blob(BlobHash::Blake3([8; 32]))).unwrap_err();
plan_asset(&store, &doc_id, &ContentKey::Blob { hash: BlobHash::Blake3([8; 32]) })
.unwrap_err();
assert!(matches!(err, AssetSyncError::UnknownAsset { .. }), "{err:?}");
}
}
+5 -4
View File
@@ -192,7 +192,7 @@ fn record_peer_presence(
found: bool,
context: &str,
) {
let key = ContentKey::Blob(hash.clone());
let key = ContentKey::Blob { hash: hash.clone() };
let evidence = &store.node_content_status_store;
let result = evidence.batch(|tx| {
if found {
@@ -602,7 +602,7 @@ mod tests {
record_peer_presence(&store, node(1), &hash(7), found, "test");
let evidence = &store.node_content_status_store;
let recorded = evidence
.batch(|tx| evidence.presence_of(tx, &node(1), &ContentKey::Blob(hash(7))))
.batch(|tx| evidence.presence_of(tx, &node(1), &ContentKey::Blob { hash: hash(7) }))
.unwrap();
assert_eq!(recorded.map(|(presence, _)| presence), Some(expected));
}
@@ -612,8 +612,9 @@ mod tests {
record_peer_presence(&store, node(1), &hash(7), true, "test");
record_peer_presence(&store, node(1), &hash(7), false, "test");
let evidence = &store.node_content_status_store;
let holders =
evidence.batch(|tx| evidence.nodes_holding(tx, &ContentKey::Blob(hash(7)))).unwrap();
let holders = evidence
.batch(|tx| evidence.nodes_holding(tx, &ContentKey::Blob { hash: hash(7) }))
.unwrap();
assert!(holders.is_empty(), "a peer that just said 'not found' is not a holder");
}
}
+16 -12
View File
@@ -894,8 +894,8 @@ fn build_content_release_batch(
let mut releases = Vec::with_capacity(released.len());
for (key, released_at) in released {
let still_here = match &key {
ContentKey::Blob(hash) => blob_store.exists(hash),
ContentKey::Artifact(id) => {
ContentKey::Blob { hash } => blob_store.exists(hash),
ContentKey::Artifact { id } => {
artifact_store.batch(|tx| artifact_store.get_artifact_by_id(tx, *id))?.is_some()
}
};
@@ -1595,7 +1595,7 @@ fn apply_records(
origin_node,
},
)?;
stored_hash.map(ContentKey::Blob)
stored_hash.map(|hash| ContentKey::Blob { hash })
} else {
artifact_store.record_artifact_stub(
tx,
@@ -1606,7 +1606,7 @@ fn apply_records(
modified_at,
origin_node,
)?;
Some(ContentKey::Artifact(Uuid::from_bytes(artifact_rec.id)))
Some(ContentKey::Artifact { id: Uuid::from_bytes(artifact_rec.id) })
};
if let Some(key) = content_key
&& let Some(has_content) = artifact_rec.has_content
@@ -1917,7 +1917,9 @@ mod tests {
let hash = BlobHash::Blake3([0x33; 32]);
let evidence = &store.node_content_status_store;
evidence
.batch(|tx| evidence.mark_present(tx, &sender, &ContentKey::Blob(hash.clone())))
.batch(|tx| {
evidence.mark_present(tx, &sender, &ContentKey::Blob { hash: hash.clone() })
})
.unwrap();
let batch = ContentReleaseBatch {
@@ -1931,7 +1933,7 @@ mod tests {
assert_eq!(
evidence
.batch(|tx| evidence.presence_of(tx, &sender, &ContentKey::Blob(hash)))
.batch(|tx| evidence.presence_of(tx, &sender, &ContentKey::Blob { hash }))
.unwrap()
.map(|(presence, _)| presence),
Some(ContentPresence::Absent),
@@ -1966,8 +1968,8 @@ mod tests {
let releases = &store.content_release_store;
releases
.batch(|tx| {
releases.record(tx, &ContentKey::Blob(gone.clone()))?;
releases.record(tx, &ContentKey::Blob(back.clone()))
releases.record(tx, &ContentKey::Blob { hash: gone.clone() })?;
releases.record(tx, &ContentKey::Blob { hash: back.clone() })
})
.unwrap();
@@ -2098,7 +2100,9 @@ mod tests {
let presence = |id: Uuid| {
node_content_store
.batch(|tx| node_content_store.presence_of(tx, &sender, &ContentKey::Artifact(id)))
.batch(|tx| {
node_content_store.presence_of(tx, &sender, &ContentKey::Artifact { id })
})
.unwrap()
.map(|(presence, _)| presence)
};
@@ -2159,7 +2163,7 @@ mod tests {
.batch(|tx| node_content_store.confirmed_at(
tx,
&sender,
&ContentKey::Blob(BlobHash::Blake3(held))
&ContentKey::Blob { hash: BlobHash::Blake3(held) }
))
.unwrap()
.is_some(),
@@ -2170,7 +2174,7 @@ mod tests {
.batch(|tx| node_content_store.confirmed_at(
tx,
&sender,
&ContentKey::Blob(BlobHash::Blake3(referenced))
&ContentKey::Blob { hash: BlobHash::Blake3(referenced) }
))
.unwrap()
.is_none(),
@@ -2181,7 +2185,7 @@ mod tests {
.batch(|tx| node_content_store.presence_of(
tx,
&sender,
&ContentKey::Blob(BlobHash::Blake3(referenced))
&ContentKey::Blob { hash: BlobHash::Blake3(referenced) }
))
.unwrap()
.map(|(presence, _)| presence),
+17 -15
View File
@@ -277,7 +277,9 @@ pub async fn sync_with_peer(
// legitimate sync, so it is logged and treated as "nothing active".
let already_running = store
.job_store
.batch(|tx| store.job_store.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer(peer)))
.batch(|tx| {
store.job_store.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer { node: peer })
})
.unwrap_or_else(|e| {
warn!("jobs: failed to check for an active sync with {label}: {e}");
None
@@ -289,7 +291,7 @@ pub async fn sync_with_peer(
let Some(handle) = job_runner.start(
JobKind::CatalogSync,
JobSubject::Peer(peer),
JobSubject::Peer { node: peer },
format!("Sync with {label}"),
origin,
) else {
@@ -498,7 +500,7 @@ pub async fn run_catalog_pass(
let Some(handle) = job_runner.start(
JobKind::CatalogSync,
JobSubject::Peer(node_pubkey),
JobSubject::Peer { node: node_pubkey },
format!("Sync with {label}"),
JobOrigin::Scheduler,
) else {
@@ -644,8 +646,8 @@ pub async fn run_catalog_fetch_pass(
// instead of whichever peer happens to come first in the list.
let wanted: Vec<ContentKey> = missing_thumbs
.iter()
.map(|h| ContentKey::Blob(BlobHash::Blake3(*h)))
.chain(pending_artifacts.iter().map(|s| ContentKey::Artifact(s.id)))
.map(|h| ContentKey::Blob { hash: BlobHash::Blake3(*h) })
.chain(pending_artifacts.iter().map(|s| ContentKey::Artifact { id: s.id }))
.collect();
let targets = peer_order_by_presence(store, peers, &wanted);
@@ -739,7 +741,7 @@ pub async fn run_content_pass(
for (doc_id, hashes) in by_document {
let Some(handle) = job_runner.start(
JobKind::ContentFetch,
JobSubject::Document(doc_id.clone()),
JobSubject::Document { document_id: doc_id.clone() },
format!("Content fetch for {doc_id}"),
JobOrigin::Scheduler,
) else {
@@ -812,7 +814,7 @@ pub(crate) fn peer_order_by_blob_presence(
hashes: &[[u8; 32]],
) -> Vec<NodePubkey> {
let keys: Vec<ContentKey> =
hashes.iter().map(|h| ContentKey::Blob(BlobHash::Blake3(*h))).collect();
hashes.iter().map(|h| ContentKey::Blob { hash: BlobHash::Blake3(*h) }).collect();
peer_order_by_presence(store, peers, &keys)
}
@@ -922,17 +924,17 @@ mod tests {
store.node_content_status_store.mark_present(
tx,
&node(0xA1),
&ContentKey::Blob(BlobHash::Blake3(hashes[0])),
&ContentKey::Blob { hash: BlobHash::Blake3(hashes[0]) },
)?;
store.node_content_status_store.mark_present(
tx,
&node(0xA1),
&ContentKey::Blob(BlobHash::Blake3(hashes[1])),
&ContentKey::Blob { hash: BlobHash::Blake3(hashes[1]) },
)?;
store.node_content_status_store.mark_present(
tx,
&node(0xB2),
&ContentKey::Blob(BlobHash::Blake3(hashes[0])),
&ContentKey::Blob { hash: BlobHash::Blake3(hashes[0]) },
)
})
.unwrap();
@@ -957,10 +959,10 @@ mod tests {
.node_content_status_store
.batch(|tx| {
let s = &store.node_content_status_store;
s.mark_present(tx, &node(0xA1), &ContentKey::Artifact(artifact))?;
s.mark_present(tx, &node(0xB2), &ContentKey::Blob(hash.clone()))?;
s.mark_present(tx, &node(0xC3), &ContentKey::Blob(hash.clone()))?;
s.mark_present(tx, &node(0xC3), &ContentKey::Artifact(artifact))
s.mark_present(tx, &node(0xA1), &ContentKey::Artifact { id: artifact })?;
s.mark_present(tx, &node(0xB2), &ContentKey::Blob { hash: hash.clone() })?;
s.mark_present(tx, &node(0xC3), &ContentKey::Blob { hash: hash.clone() })?;
s.mark_present(tx, &node(0xC3), &ContentKey::Artifact { id: artifact })
})
.unwrap();
@@ -968,7 +970,7 @@ mod tests {
let order = peer_order_by_presence(
&store,
&peers,
&[ContentKey::Blob(hash), ContentKey::Artifact(artifact)],
&[ContentKey::Blob { hash }, ContentKey::Artifact { id: artifact }],
);
assert_eq!(
+1 -1
View File
@@ -146,7 +146,7 @@ mod tests {
use rstest::rstest;
fn blob(b: u8) -> ContentKey {
ContentKey::Blob(BlobHash::Blake3([b; 32]))
ContentKey::Blob { hash: BlobHash::Blake3([b; 32]) }
}
#[rstest]
+25 -19
View File
@@ -315,7 +315,7 @@ mod tests {
#[rstest]
fn create_and_get_roundtrips(store: TestStore) {
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(1)));
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(1) });
let s = &store.job_store;
let fetched = s
.batch(|tx| {
@@ -326,13 +326,13 @@ mod tests {
.unwrap();
assert_eq!(fetched.id, job.id);
assert_eq!(fetched.kind, JobKind::CatalogSync);
assert_eq!(fetched.subject, JobSubject::Peer(node(1)));
assert_eq!(fetched.subject, JobSubject::Peer { node: node(1) });
assert_eq!(fetched.state, JobState::Running);
}
#[rstest]
fn set_progress_roundtrips(store: TestStore) {
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(1)));
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(1) });
let s = &store.job_store;
let progress = JobProgress {
units_done: 3,
@@ -354,8 +354,10 @@ mod tests {
#[rstest]
fn finish_sets_terminal_state_and_error(store: TestStore) {
let job =
sample_job(JobKind::Push, JobSubject::Document(crate::documents::DocumentID::new()));
let job = sample_job(
JobKind::Push,
JobSubject::Document { document_id: crate::documents::DocumentID::new() },
);
let s = &store.job_store;
let fetched = s
.batch(|tx| {
@@ -371,8 +373,8 @@ mod tests {
#[rstest]
fn list_active_and_history_partition_correctly(store: TestStore) {
let running = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(1)));
let mut done = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(2)));
let running = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(1) });
let mut done = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(2) });
done.state = JobState::Done;
let s = &store.job_store;
let (active, history, count) = s
@@ -401,16 +403,16 @@ mod tests {
#[rstest]
fn find_active_matches_kind_and_subject(store: TestStore) {
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(1)));
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(1) });
let s = &store.job_store;
let (same, diff_subject, diff_kind) = s
.batch(|tx| {
s.create(tx, &job)?;
Ok::<_, JobStoreError>((
s.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer(node(1)))?,
s.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer(node(2)))?,
s.find_active(tx, JobKind::Push, &JobSubject::Peer(node(1)))?,
s.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer { node: node(1) })?,
s.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer { node: node(2) })?,
s.find_active(tx, JobKind::Push, &JobSubject::Peer { node: node(1) })?,
))
})
.unwrap();
@@ -424,7 +426,7 @@ mod tests {
let after_finish = s
.batch(|tx| {
s.finish(tx, job.id, JobState::Done)?;
s.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer(node(1)))
s.find_active(tx, JobKind::CatalogSync, &JobSubject::Peer { node: node(1) })
})
.unwrap();
assert_eq!(after_finish, None);
@@ -433,7 +435,10 @@ mod tests {
#[rstest]
fn document_to_peer_subject_distinguishes_peers(store: TestStore) {
let doc_id = crate::documents::DocumentID::new();
let job = sample_job(JobKind::Push, JobSubject::DocumentToPeer(doc_id.clone(), node(1)));
let job = sample_job(
JobKind::Push,
JobSubject::DocumentToPeer { document_id: doc_id.clone(), node: node(1) },
);
let s = &store.job_store;
// Same document, different peer: not a match — the whole point of
@@ -446,12 +451,12 @@ mod tests {
s.find_active(
tx,
JobKind::Push,
&JobSubject::DocumentToPeer(doc_id.clone(), node(2)),
&JobSubject::DocumentToPeer { document_id: doc_id.clone(), node: node(2) },
)?,
s.find_active(
tx,
JobKind::Push,
&JobSubject::DocumentToPeer(doc_id.clone(), node(1)),
&JobSubject::DocumentToPeer { document_id: doc_id.clone(), node: node(1) },
)?,
))
})
@@ -462,7 +467,7 @@ mod tests {
#[rstest]
fn reconcile_orphaned_running_marks_stale_rows_failed(store: TestStore) {
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(1)));
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(1) });
let s = &store.job_store;
let (n, fetched) = s
.batch(|tx| {
@@ -487,7 +492,8 @@ mod tests {
let history_len = s
.batch(|tx| {
for i in 0..5u8 {
let mut job = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(i)));
let mut job =
sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(i) });
job.id = Uuid::new_v4();
s.create(tx, &job)?;
s.finish(tx, job.id, JobState::Done)?;
@@ -501,8 +507,8 @@ mod tests {
#[rstest]
fn clear_history_deletes_terminal_jobs_only(store: TestStore) {
let active = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(1)));
let done = sample_job(JobKind::CatalogSync, JobSubject::Peer(node(2)));
let active = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(1) });
let done = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: node(2) });
let s = &store.job_store;
let (n, history, active_count) = s
+21 -13
View File
@@ -52,13 +52,17 @@ crate::stores::rusqlite_error_from!(NodeContentStatusError, Sqlite);
/// *bytes* and evidence gathered for one artifact answers for every other
/// artifact with the same hash. An inline value lives in its artifact's row
/// and has no identity apart from it.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
///
/// Also how a frontend names one artifact to sync on its own
/// ([`EnsembleFfi::sync_document_asset`](crate::ffi::EnsembleFfi::sync_document_asset));
/// it already holds both halves from the rows it renders.
#[derive(Debug, Clone, PartialEq, Eq, Hash, uniffi::Enum)]
pub enum ContentKey {
/// Ciphertext bytes in the content-addressed blob store, behind a
/// [`BlobRef`](crate::documents::BlobRef).
Blob(BlobHash),
Blob { hash: BlobHash },
/// One inline artifact's value, identified by the artifact's id.
Artifact(Uuid),
Artifact { id: Uuid },
}
impl ContentKey {
@@ -67,19 +71,23 @@ impl ContentKey {
/// table the same way.
pub(crate) fn columns(&self) -> (&'static str, String) {
match self {
ContentKey::Blob(hash) => (KIND_BLOB, hash.to_hex()),
ContentKey::Artifact(id) => (KIND_ARTIFACT, id.to_string()),
ContentKey::Blob { hash } => (KIND_BLOB, hash.to_hex()),
ContentKey::Artifact { id } => (KIND_ARTIFACT, id.to_string()),
}
}
pub(crate) fn from_columns(kind: &str, id: &str) -> Result<Self> {
match kind {
KIND_BLOB => BlobHash::from_blake3_hex(id).map(ContentKey::Blob).map_err(|e| {
NodeContentStatusError::InvalidContentId(id.to_string(), e.to_string())
}),
KIND_ARTIFACT => Uuid::parse_str(id).map(ContentKey::Artifact).map_err(|e| {
NodeContentStatusError::InvalidContentId(id.to_string(), e.to_string())
}),
KIND_BLOB => {
BlobHash::from_blake3_hex(id).map(|hash| ContentKey::Blob { hash }).map_err(|e| {
NodeContentStatusError::InvalidContentId(id.to_string(), e.to_string())
})
}
KIND_ARTIFACT => {
Uuid::parse_str(id).map(|id| ContentKey::Artifact { id }).map_err(|e| {
NodeContentStatusError::InvalidContentId(id.to_string(), e.to_string())
})
}
other => Err(NodeContentStatusError::UnknownContentKind(other.to_string())),
}
}
@@ -347,11 +355,11 @@ mod tests {
use rstest::rstest;
fn blob(b: u8) -> ContentKey {
ContentKey::Blob(BlobHash::Blake3([b; 32]))
ContentKey::Blob { hash: BlobHash::Blake3([b; 32]) }
}
fn artifact(b: u8) -> ContentKey {
ContentKey::Artifact(Uuid::from_bytes([b; 16]))
ContentKey::Artifact { id: Uuid::from_bytes([b; 16]) }
}
#[rstest]
+19 -13
View File
@@ -276,7 +276,7 @@ fn document_detail_snapshot(store: &Store, id: &DocumentID) -> Result<DocumentDe
blob_refs
.iter()
.map(|(kind, blob)| {
let key = ContentKey::Blob(blob.blob_hash.clone());
let key = ContentKey::Blob { hash: blob.blob_hash.clone() };
let released_at =
store.content_release_store.released_at(tx, &key).ok().flatten();
let nodes = doc_nodes
@@ -382,7 +382,7 @@ fn document_detail_snapshot(store: &Store, id: &DocumentID) -> Result<DocumentDe
// Either `art` or `stub` is Some by the time we get here; both carry
// the artifact id the evidence store is keyed on.
let evidence_key =
art.map(|a| a.id).or_else(|| stub.map(|s| s.id)).map(ContentKey::Artifact);
art.map(|a| a.id).or_else(|| stub.map(|s| s.id)).map(|id| ContentKey::Artifact { id });
let mut nodes: Vec<NodePresenceRow> = store.node_content_status_store.batch(|tx| {
Ok::<_, synchronicity::stores::node_content_status::NodeContentStatusError>(
doc_nodes
@@ -496,7 +496,7 @@ fn import_paths(
run_as_job_with_progress(
job_runner,
JobKind::Import,
JobSubject::Path(label.clone()),
JobSubject::Path { path: label.clone() },
label,
|reporter| -> Result<ImportOutcome> {
let progress = ImportProgress::files(Some(reporter), Some(totals));
@@ -764,7 +764,7 @@ async fn import_url(store: &Store, job_runner: &JobRunner, url: &str) -> Result<
run_as_job_async_with_progress(
job_runner,
JobKind::Import,
JobSubject::Path(url.to_string()),
JobSubject::Path { path: url.to_string() },
url.to_string(),
|reporter| async move {
let progress =
@@ -1453,10 +1453,15 @@ fn job_snapshot(
peer_names: &std::collections::HashMap<String, String>,
) -> JobSnapshot {
let (peer_pubkey, document_id) = match &job.subject {
JobSubject::Peer(pk) => (Some(pk.to_hex()), None),
JobSubject::Document(id) | JobSubject::DocumentAsset(id, _) => (None, Some(id.clone())),
JobSubject::DocumentToPeer(id, pk) => (Some(pk.to_hex()), Some(id.clone())),
JobSubject::DocumentType(_) | JobSubject::Path(_) | JobSubject::None => (None, None),
JobSubject::Peer { node: pk } => (Some(pk.to_hex()), None),
JobSubject::Document { document_id: id }
| JobSubject::DocumentAsset { document_id: id, .. } => (None, Some(id.clone())),
JobSubject::DocumentToPeer { document_id: id, node: pk } => {
(Some(pk.to_hex()), Some(id.clone()))
}
JobSubject::DocumentType { .. } | JobSubject::Path { .. } | JobSubject::None => {
(None, None)
}
};
let peer_name = peer_pubkey.as_ref().and_then(|hex| peer_names.get(hex).cloned());
let document_title = document_id
@@ -1753,7 +1758,7 @@ async fn dispatch_sync_asset(
let res = run_as_job_async_with_progress(
job_runner,
JobKind::ContentFetch,
JobSubject::DocumentAsset(id.clone(), asset.clone()),
JobSubject::DocumentAsset { document_id: id.clone(), asset: asset.clone() },
format!("Pull {asset} of {id}"),
// Mapped to `anyhow` inside the job rather than by the `?` outside it:
// the wrapper needs one error type that can also carry an
@@ -1807,7 +1812,7 @@ async fn dispatch_sync_assets(
let res = run_as_job_async_with_progress(
job_runner,
JobKind::ContentFetch,
JobSubject::Document(id.clone()),
JobSubject::Document { document_id: id.clone() },
format!("Pull {id}"),
|reporter| async move {
synchronicity::p2p::asset_sync::sync_document_assets_recursive(
@@ -1996,7 +2001,8 @@ mod job_snapshot_tests {
})
.unwrap();
let job = sample_job(JobKind::ContentFetch, JobSubject::Document(doc_id.clone()));
let job =
sample_job(JobKind::ContentFetch, JobSubject::Document { document_id: doc_id.clone() });
let snapshot = job_snapshot(&store, &job, &std::collections::HashMap::new());
assert_eq!(snapshot.document_id, Some(doc_id.clone()));
@@ -2017,7 +2023,7 @@ mod job_snapshot_tests {
let store = Store::create(dir.path()).unwrap();
let peer = NodePubkey([2u8; 32]);
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer(peer));
let job = sample_job(JobKind::CatalogSync, JobSubject::Peer { node: peer });
let mut names = std::collections::HashMap::new();
names.insert(peer.to_hex(), "Greg's Laptop".to_string());
@@ -2036,7 +2042,7 @@ mod job_snapshot_tests {
fn failed_state_carries_error_message() {
let dir = TempDir::new().unwrap();
let store = Store::create(dir.path()).unwrap();
let mut job = sample_job(JobKind::Import, JobSubject::Path("a.pdf".to_string()));
let mut job = sample_job(JobKind::Import, JobSubject::Path { path: "a.pdf".to_string() });
job.state = JobState::Failed { message: "disk full".to_string() };
let snapshot = job_snapshot(&store, &job, &std::collections::HashMap::new());