mirror of
https://github.com/GothenburgBitFactory/taskwarrior.git
synced 2025-07-07 20:06:36 +02:00
use log and env_logger, and add some logging around sync
This commit is contained in:
parent
c117494ce6
commit
0a1ee470f7
13 changed files with 109 additions and 14 deletions
|
@ -13,6 +13,7 @@ failure = {version = "^0.1.5", features = ["derive"] }
|
|||
kv = {version = "^0.10.0", features = ["msgpack-value"]}
|
||||
lmdb-rkv = {version = "^0.12.3"}
|
||||
ureq = "^1.5.2"
|
||||
log = "^0.4.11"
|
||||
|
||||
[dev-dependencies]
|
||||
proptest = "^0.9.4"
|
||||
|
|
|
@ -6,6 +6,7 @@ use crate::taskdb::TaskDB;
|
|||
use crate::taskstorage::{KVStorage, Operation, TaskMap, TaskStorage};
|
||||
use chrono::Utc;
|
||||
use failure::Fallible;
|
||||
use log::trace;
|
||||
use std::collections::HashMap;
|
||||
use uuid::Uuid;
|
||||
|
||||
|
@ -128,6 +129,7 @@ impl Replica {
|
|||
pub fn new_task(&mut self, status: Status, description: String) -> Fallible<Task> {
|
||||
let uuid = Uuid::new_v4();
|
||||
self.taskdb.apply(Operation::Create { uuid })?;
|
||||
trace!("task {} created", uuid);
|
||||
let mut task = Task::new(uuid, TaskMap::new()).into_mut(self);
|
||||
task.set_description(description)?;
|
||||
task.set_status(status)?;
|
||||
|
@ -144,7 +146,9 @@ impl Replica {
|
|||
if self.taskdb.get_task(&uuid)?.is_none() {
|
||||
return Err(Error::DBError(format!("Task {} does not exist", uuid)).into());
|
||||
}
|
||||
self.taskdb.apply(Operation::Delete { uuid: *uuid })
|
||||
self.taskdb.apply(Operation::Delete { uuid: *uuid })?;
|
||||
trace!("task {} deleted", uuid);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Synchronize this replica against the given server.
|
||||
|
|
|
@ -2,6 +2,7 @@ use crate::replica::Replica;
|
|||
use crate::taskstorage::TaskMap;
|
||||
use chrono::prelude::*;
|
||||
use failure::Fallible;
|
||||
use log::trace;
|
||||
use uuid::Uuid;
|
||||
|
||||
pub type Timestamp = DateTime<Utc>;
|
||||
|
@ -230,6 +231,7 @@ impl<'r> TaskMut<'r> {
|
|||
let now = format!("{}", Utc::now().timestamp());
|
||||
self.replica
|
||||
.update_task(self.task.uuid, "modified", Some(now.clone()))?;
|
||||
trace!("task {}: set property modified={:?}", self.task.uuid, now);
|
||||
self.task.taskmap.insert(String::from("modified"), now);
|
||||
self.updated_modified = true;
|
||||
}
|
||||
|
@ -242,8 +244,10 @@ impl<'r> TaskMut<'r> {
|
|||
.update_task(self.task.uuid, property, value.as_ref())?;
|
||||
|
||||
if let Some(v) = value {
|
||||
trace!("task {}: set property {}={:?}", self.task.uuid, property, v);
|
||||
self.task.taskmap.insert(property.to_string(), v);
|
||||
} else {
|
||||
trace!("task {}: remove property {}", self.task.uuid, property);
|
||||
self.task.taskmap.remove(property);
|
||||
}
|
||||
Ok(())
|
||||
|
|
|
@ -2,6 +2,7 @@ use crate::errors::Error;
|
|||
use crate::server::{AddVersionResult, GetVersionResult, Server};
|
||||
use crate::taskstorage::{Operation, TaskMap, TaskStorage, TaskStorageTxn};
|
||||
use failure::{format_err, Fallible};
|
||||
use log::{info, trace, warn};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashSet;
|
||||
use std::str;
|
||||
|
@ -172,10 +173,12 @@ impl TaskDB {
|
|||
// version twice, then we have diverged.
|
||||
let mut requested_parent_version_id = None;
|
||||
loop {
|
||||
trace!("beginning sync outer loop");
|
||||
let mut base_version_id = txn.base_version()?;
|
||||
|
||||
// first pull changes and "rebase" on top of them
|
||||
loop {
|
||||
trace!("beginning sync inner loop");
|
||||
if let GetVersionResult::Version {
|
||||
version_id,
|
||||
history_segment,
|
||||
|
@ -184,14 +187,14 @@ impl TaskDB {
|
|||
{
|
||||
let version_str = str::from_utf8(&history_segment).unwrap();
|
||||
let version: Version = serde_json::from_str(version_str).unwrap();
|
||||
println!("applying version {:?} from server", version_id);
|
||||
|
||||
// apply this verison and update base_version in storage
|
||||
info!("applying version {:?} from server", version_id);
|
||||
TaskDB::apply_version(txn.as_mut(), version)?;
|
||||
txn.set_base_version(version_id)?;
|
||||
base_version_id = version_id;
|
||||
} else {
|
||||
println!("no child versions of {:?}", base_version_id);
|
||||
info!("no child versions of {:?}", base_version_id);
|
||||
// at the moment, no more child versions, so we can try adding our own
|
||||
break;
|
||||
}
|
||||
|
@ -199,24 +202,26 @@ impl TaskDB {
|
|||
|
||||
let operations: Vec<Operation> = txn.operations()?.to_vec();
|
||||
if operations.is_empty() {
|
||||
println!("no changes to push to server");
|
||||
info!("no changes to push to server");
|
||||
// nothing to sync back to the server..
|
||||
break;
|
||||
}
|
||||
|
||||
trace!("sending {} operations to the server", operations.len());
|
||||
|
||||
// now make a version of our local changes and push those
|
||||
let new_version = Version { operations };
|
||||
let history_segment = serde_json::to_string(&new_version).unwrap().into();
|
||||
println!("sending new version to server");
|
||||
info!("sending new version to server");
|
||||
match server.add_version(base_version_id, history_segment)? {
|
||||
AddVersionResult::Ok(new_version_id) => {
|
||||
println!("version {:?} received by server", new_version_id);
|
||||
info!("version {:?} received by server", new_version_id);
|
||||
txn.set_base_version(new_version_id)?;
|
||||
txn.set_operations(vec![])?;
|
||||
break;
|
||||
}
|
||||
AddVersionResult::ExpectedParentVersion(parent_version_id) => {
|
||||
println!(
|
||||
info!(
|
||||
"new version rejected; must be based on {:?}",
|
||||
parent_version_id
|
||||
);
|
||||
|
@ -264,22 +269,31 @@ impl TaskDB {
|
|||
// it. If it happens for server op, then we must copy the remaining local ops.
|
||||
let mut local_operations: Vec<Operation> = txn.operations()?;
|
||||
for server_op in version.operations.drain(..) {
|
||||
trace!(
|
||||
"rebasing local operations onto server operation {:?}",
|
||||
server_op
|
||||
);
|
||||
let mut new_local_ops = Vec::with_capacity(local_operations.len());
|
||||
let mut svr_op = Some(server_op);
|
||||
for local_op in local_operations.drain(..) {
|
||||
if let Some(o) = svr_op {
|
||||
let (new_server_op, new_local_op) = Operation::transform(o, local_op);
|
||||
let (new_server_op, new_local_op) = Operation::transform(o, local_op.clone());
|
||||
trace!("local operation {:?} -> {:?}", local_op, new_local_op);
|
||||
svr_op = new_server_op;
|
||||
if let Some(o) = new_local_op {
|
||||
new_local_ops.push(o);
|
||||
}
|
||||
} else {
|
||||
trace!(
|
||||
"local operation {:?} unchanged (server operation consumed)",
|
||||
local_op
|
||||
);
|
||||
new_local_ops.push(local_op);
|
||||
}
|
||||
}
|
||||
if let Some(o) = svr_op {
|
||||
if let Err(e) = TaskDB::apply_op(txn, &o) {
|
||||
println!("Invalid operation when syncing: {} (ignored)", e);
|
||||
warn!("Invalid operation when syncing: {} (ignored)", e);
|
||||
}
|
||||
}
|
||||
local_operations = new_local_ops;
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue