rgw: track object versions between reads and writes

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

10 files changed, 200 insertions, 89 deletions

diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 9834c0c851d..246d0d528b6 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -872,6 +872,7 @@ int main(int argc, char **argv)
   int check_objects = false;
   string infile;
   string metadata_key;
+  RGWObjVersionTracker objv_tracker;
 
   std::string val;
   std::ostringstream errs;
@@ -1286,7 +1287,7 @@ int main(int argc, char **argv)
     string s;
     if (!found && (!user_email.empty())) {
       s = user_email;
-      if (rgw_get_user_info_by_email(store, s, info) >= 0) {
+      if (rgw_get_user_info_by_email(store, s, info, &objv_tracker) >= 0) {
 	found = true;
       } else {
 	cerr << "could not find user by specified email" << std::endl;
@@ -1294,7 +1295,7 @@ int main(int argc, char **argv)
     }
     if (!found && (!access_key.empty())) {
       s = access_key;
-      if (rgw_get_user_info_by_access_key(store, s, info) >= 0) {
+      if (rgw_get_user_info_by_access_key(store, s, info, &objv_tracker) >= 0) {
 	found = true;
       } else {
 	cerr << "could not find user by specified access key" << std::endl;
@@ -1313,7 +1314,7 @@ int main(int argc, char **argv)
       return usage();
     }
 
-    bool found = (rgw_get_user_info_by_uid(store, user_id, info) >= 0);
+    bool found = (rgw_get_user_info_by_uid(store, user_id, info, &objv_tracker) >= 0);
 
     if (opt_cmd == OPT_USER_CREATE) {
       if (found) {
@@ -1395,7 +1396,7 @@ int main(int argc, char **argv)
 	}
 	access_key = public_id_buf;
 	duplicate_check_id = access_key;
-      } while (!rgw_get_user_info_by_access_key(store, duplicate_check_id, duplicate_check));
+      } while (!rgw_get_user_info_by_access_key(store, duplicate_check_id, duplicate_check, &objv_tracker));
     }
   }
 
diff --git a/src/rgw/rgw_cache.h b/src/rgw/rgw_cache.h
index 2364aab3e4f..ec09552b524 100644
--- a/src/rgw/rgw_cache.h
+++ b/src/rgw/rgw_cache.h
@@ -193,13 +193,14 @@ public:
                    map<std::string, bufferlist>& attrs, RGWObjCategory category, int flags,
                    map<std::string, bufferlist>* rmattrs, const bufferlist *data,
                    RGWObjManifest *manifest, const string *ptag, list<string> *remove_objs,
-                   bool modify_version, obj_version *objv);
+                   bool modify_version, RGWObjVersionTracker *objv_tracker);
   int put_obj_data(void *ctx, rgw_obj& obj, const char *data,
               off_t ofs, size_t len, bool exclusive);
 
-  int get_obj(void *ctx, obj_version *objv, void **handle, rgw_obj& obj, bufferlist& bl, off_t ofs, off_t end);
+  int get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **handle, rgw_obj& obj, bufferlist& bl, off_t ofs, off_t end);
 
-  int obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime, uint64_t *epoch, map<string, bufferlist> *attrs, bufferlist *first_chunk, obj_version *objv);
+  int obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime, uint64_t *epoch, map<string, bufferlist> *attrs,
+               bufferlist *first_chunk, RGWObjVersionTracker *objv_tracker);
 
   int delete_obj(void *ctx, rgw_obj& obj);
 };
@@ -235,13 +236,13 @@ int RGWCache<T>::delete_obj(void *ctx, rgw_obj& obj)
 }
 
 template <class T>
-int RGWCache<T>::get_obj(void *ctx, obj_version *objv, void **handle, rgw_obj& obj, bufferlist& obl, off_t ofs, off_t end)
+int RGWCache<T>::get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **handle, rgw_obj& obj, bufferlist& obl, off_t ofs, off_t end)
 {
   rgw_bucket bucket;
   string oid;
   normalize_bucket_and_obj(obj.bucket, obj.object, bucket, oid);
   if (bucket.name[0] != '.' || ofs != 0)
-    return T::get_obj(ctx, objv, handle, obj, obl, ofs, end);
+    return T::get_obj(ctx, objv_tracker, handle, obj, obl, ofs, end);
 
   string name = normal_name(obj.bucket, oid);
 
@@ -259,7 +260,7 @@ int RGWCache<T>::get_obj(void *ctx, obj_version *objv, void **handle, rgw_obj& o
     i.copy_all(obl);
     return bl.length();
   }
-  int r = T::get_obj(ctx, objv, handle, obj, obl, ofs, end);
+  int r = T::get_obj(ctx, objv_tracker, handle, obj, obl, ofs, end);
   if (r < 0) {
     if (r == -ENOENT) { // only update ENOENT, we'd rather retry other errors
       info.status = r;
@@ -353,7 +354,7 @@ int RGWCache<T>::put_obj_meta_impl(void *ctx, rgw_obj& obj, uint64_t size, time_
                               map<std::string, bufferlist>& attrs, RGWObjCategory category, int flags,
                               map<std::string, bufferlist>* rmattrs, const bufferlist *data,
                               RGWObjManifest *manifest, const string *ptag, list<string> *remove_objs,
-                              bool modify_version, obj_version *objv)
+                              bool modify_version, RGWObjVersionTracker *objv_tracker)
 {
   rgw_bucket bucket;
   string oid;
@@ -371,7 +372,7 @@ int RGWCache<T>::put_obj_meta_impl(void *ctx, rgw_obj& obj, uint64_t size, time_
     }
   }
   int ret = T::put_obj_meta_impl(ctx, obj, size, mtime, attrs, category, flags, rmattrs, data, manifest, ptag, remove_objs,
-                                 modify_version, objv);
+                                 modify_version, objv_tracker);
   if (cacheable) {
     string name = normal_name(bucket, oid);
     if (ret >= 0) {
@@ -425,13 +426,13 @@ int RGWCache<T>::put_obj_data(void *ctx, rgw_obj& obj, const char *data,
 template <class T>
 int RGWCache<T>::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime,
                           uint64_t *pepoch, map<string, bufferlist> *attrs,
-                          bufferlist *first_chunk, obj_version *objv)
+                          bufferlist *first_chunk, RGWObjVersionTracker *objv_tracker)
 {
   rgw_bucket bucket;
   string oid;
   normalize_bucket_and_obj(obj.bucket, obj.object, bucket, oid);
   if (bucket.name[0] != '.')
-    return T::obj_stat(ctx, obj, psize, pmtime, pepoch, attrs, first_chunk, objv);
+    return T::obj_stat(ctx, obj, psize, pmtime, pepoch, attrs, first_chunk, objv_tracker);
 
   string name = normal_name(bucket, oid);
 
@@ -450,7 +451,7 @@ int RGWCache<T>::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmti
     epoch = info.epoch;
     goto done;
   }
-  r = T::obj_stat(ctx, obj, &size, &mtime, &epoch, &info.xattrs, first_chunk, objv);
+  r = T::obj_stat(ctx, obj, &size, &mtime, &epoch, &info.xattrs, first_chunk, objv_tracker);
   if (r < 0) {
     if (r == -ENOENT) {
       info.status = r;
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index 9296dc28e78..a1fc1512751 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -24,7 +24,9 @@ public:
   virtual string get_type() { return string(); }
 
   virtual int get(RGWRados *store, string& entry, RGWMetadataObject **obj) { return -ENOTSUP; }
-  virtual int put(RGWRados *store, string& entry, obj_version& objv, JSONObj *obj) { return -ENOTSUP; }
+  virtual int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, JSONObj *obj) { return -ENOTSUP; }
+  virtual int put_obj(RGWRados *store, string& key, bufferlist& bl, bool exclusive,
+                      RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs) { return -ENOTSUP; }
 
   virtual int list_keys_init(RGWRados *store, void **phandle) {
     iter_data *data = new iter_data;
@@ -77,6 +79,15 @@ int RGWMetadataManager::register_handler(RGWMetadataHandler *handler)
   return 0;
 }
 
+RGWMetadataHandler *RGWMetadataManager::get_handler(const char *type)
+{
+  map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
+  if (iter == handlers.end())
+    return NULL;
+
+  return iter->second;
+}
+
 void RGWMetadataManager::parse_metadata_key(const string& metadata_key, string& type, string& entry)
 {
   int pos = metadata_key.find(':');
@@ -151,17 +162,20 @@ int RGWMetadataManager::put(string& metadata_key, bufferlist& bl)
   }
 
   string meadata_key;
-  obj_version objv;
+  RGWObjVersionTracker objv_tracker;
+
+  obj_version *objv = &objv_tracker.write_version;
+
 
   JSONDecoder::decode_json("key", metadata_key, &parser);
-  JSONDecoder::decode_json("ver", objv, &parser);
+  JSONDecoder::decode_json("ver", *objv, &parser);
 
   JSONObj *jo = parser.find_obj("data");
   if (!jo) {
     return -EINVAL;
   }
 
-  return handler->put(store, entry, objv, jo);
+  return handler->put(store, entry, objv_tracker, jo);
 }
 
 
@@ -224,3 +238,9 @@ void RGWMetadataManager::get_sections(list<string>& sections)
 }
 
 
+int RGWMetadataManager::put_obj(RGWMetadataHandler *handler, string& key, bufferlist& bl, bool exclusive,
+                                RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs)
+{
+  return handler->put_obj(store, key, bl, exclusive, objv_tracker, pattrs);
+}
+
diff --git a/src/rgw/rgw_metadata.h b/src/rgw/rgw_metadata.h
index eb052e7b03c..d432447e43f 100644
--- a/src/rgw/rgw_metadata.h
+++ b/src/rgw/rgw_metadata.h
@@ -10,6 +10,7 @@
 
 class RGWRados;
 class JSONObj;
+class RGWObjVersionTracker;
 
 struct obj_version;
 
@@ -25,13 +26,21 @@ public:
   virtual void dump(Formatter *f) const = 0;
 };
 
+class RGWMetadataManager;
+
 class RGWMetadataHandler {
+  friend class RGWMetadataManager;
+
+protected:
+  virtual int put_obj(RGWRados *store, string& key, bufferlist& bl, bool exclusive,
+                      RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs = NULL) = 0;
 public:
   virtual ~RGWMetadataHandler() {}
   virtual string get_type() = 0;
 
   virtual int get(RGWRados *store, string& entry, RGWMetadataObject **obj) = 0;
-  virtual int put(RGWRados *store, string& entry, obj_version& objv, JSONObj *obj) = 0;
+  virtual int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, JSONObj *obj) = 0;
+
 
   virtual int list_keys_init(RGWRados *store, void **phandle) = 0;
   virtual int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) = 0;
@@ -52,6 +61,10 @@ public:
 
   int register_handler(RGWMetadataHandler *handler);
 
+  RGWMetadataHandler *get_handler(const char *type);
+
+  int put_obj(RGWMetadataHandler *handler, string& key, bufferlist& bl, bool exclusive,
+              RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs = NULL);
   int get(string& metadata_key, Formatter *f);
   int put(string& metadata_key, bufferlist& bl);
 
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index ae373775657..faa87e3bfe0 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -418,6 +418,34 @@ int RGWRegionMap::update(RGWRegion& region)
   return 0;
 }
 
+
+void RGWObjVersionTracker::prepare_op_for_read(ObjectReadOperation *op)
+{
+  obj_version *check_objv = version_for_check();
+
+  if (check_objv) {
+    cls_version_check(*op, *check_objv, VER_COND_EQ);
+  }
+
+  cls_version_read(*op, &read_version);
+}
+
+void RGWObjVersionTracker::prepare_op_for_write(ObjectWriteOperation *op)
+{
+  obj_version *check_objv = version_for_check();
+  obj_version *modify_version = version_for_write();
+
+  if (check_objv) {
+    cls_version_check(*op, *check_objv, VER_COND_EQ);
+  }
+
+  if (modify_version) {
+    cls_version_set(*op, *modify_version);
+  } else {
+    cls_version_inc(*op);
+  }
+}
+
 void RGWObjManifest::append(RGWObjManifest& m)
 {
   map<uint64_t, RGWObjManifestPart>::iterator iter;
@@ -1417,7 +1445,7 @@ int RGWRados::put_obj_meta_impl(void *ctx, rgw_obj& obj,  uint64_t size,
 		  const string *ptag,
                   list<string> *remove_objs,
                   bool modify_version,
-                  obj_version *objv)
+                  RGWObjVersionTracker *objv_tracker)
 {
   rgw_bucket bucket;
   std::string oid, key;
@@ -1446,12 +1474,8 @@ int RGWRados::put_obj_meta_impl(void *ctx, rgw_obj& obj,  uint64_t size,
       return r;
   }
 
-  if (modify_version) {
-    if (objv) {
-      cls_version_set(op, *objv);
-    } else {
-      cls_version_inc(op);
-    }
+  if (objv_tracker) {
+    objv_tracker->prepare_op_for_write(&op);
   }
 
   if (data) {
@@ -1515,6 +1539,10 @@ int RGWRados::put_obj_meta_impl(void *ctx, rgw_obj& obj,  uint64_t size,
   if (r < 0)
     goto done_cancel;
 
+  if (objv_tracker) {
+    objv_tracker->apply_write();
+  }
+
   epoch = io_ctx.get_last_version();
 
   r = complete_atomic_overwrite(rctx, state, obj);
@@ -2220,7 +2248,7 @@ static void generate_fake_tag(CephContext *cct, map<string, bufferlist>& attrset
   tag_bl.append(tag.c_str(), tag.size() + 1);
 }
 
-int RGWRados::get_obj_state(RGWRadosCtx *rctx, rgw_obj& obj, RGWObjState **state, obj_version *objv)
+int RGWRados::get_obj_state(RGWRadosCtx *rctx, rgw_obj& obj, RGWObjState **state, RGWObjVersionTracker *objv_tracker)
 {
   RGWObjState *s = rctx->get_state(obj);
   ldout(cct, 20) << "get_obj_state: rctx=" << (void *)rctx << " obj=" << obj << " state=" << (void *)s << " s->prefetch_data=" << s->prefetch_data << dendl;
@@ -2228,7 +2256,7 @@ int RGWRados::get_obj_state(RGWRadosCtx *rctx, rgw_obj& obj, RGWObjState **state
   if (s->has_attrs)
     return 0;
 
-  int r = obj_stat(rctx, obj, &s->size, &s->mtime, &s->epoch, &s->attrset, (s->prefetch_data ? &s->data : NULL), objv);
+  int r = obj_stat(rctx, obj, &s->size, &s->mtime, &s->epoch, &s->attrset, (s->prefetch_data ? &s->data : NULL), objv_tracker);
   if (r == -ENOENT) {
     s->exists = false;
     s->has_attrs = true;
@@ -2623,7 +2651,7 @@ int RGWRados::prepare_get_obj(void *ctx, rgw_obj& obj,
             const char *if_nomatch,
             uint64_t *total_size,
             uint64_t *obj_size,
-            obj_version *objv,
+            RGWObjVersionTracker *objv_tracker,
             void **handle,
             struct rgw_err *err)
 {
@@ -2660,7 +2688,7 @@ int RGWRados::prepare_get_obj(void *ctx, rgw_obj& obj,
     rctx = new_ctx;
   }
 
-  r = get_obj_state(rctx, obj, &astate, objv);
+  r = get_obj_state(rctx, obj, &astate, objv_tracker);
   if (r < 0)
     goto done_err;
 
@@ -2964,7 +2992,7 @@ int RGWRados::clone_objs(void *ctx, rgw_obj& dst_obj,
 }
 
 
-int RGWRados::get_obj(void *ctx, obj_version *objv, void **handle, rgw_obj& obj,
+int RGWRados::get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **handle, rgw_obj& obj,
                       bufferlist& bl, off_t ofs, off_t end)
 {
   rgw_bucket bucket;
@@ -3034,8 +3062,8 @@ int RGWRados::get_obj(void *ctx, obj_version *objv, void **handle, rgw_obj& obj,
       goto done_ret;
   }
 
-  if (objv) {
-    cls_version_check(op, *objv, VER_COND_EQ);
+  if (objv_tracker) {
+    objv_tracker->prepare_op_for_read(&op);
   }
 
   read_len = len;
@@ -3542,7 +3570,7 @@ int RGWRados::read(void *ctx, rgw_obj& obj, off_t ofs, size_t size, bufferlist&
 }
 
 int RGWRados::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime, uint64_t *epoch, map<string, bufferlist> *attrs, bufferlist *first_chunk,
-                       obj_version *objv)
+                       RGWObjVersionTracker *objv_tracker)
 {
   rgw_bucket bucket;
   std::string oid, key;
@@ -3559,8 +3587,8 @@ int RGWRados::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime,
   time_t mtime = 0;
 
   ObjectReadOperation op;
-  if (objv) {
-    cls_version_read(op, objv);
+  if (objv_tracker) {
+    objv_tracker->prepare_op_for_read(&op);
   }
   op.getxattrs(&attrset, NULL);
   op.stat(&size, &mtime, NULL);
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index f40ef180ef1..a518574927a 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -41,6 +41,37 @@ static inline void get_obj_bucket_and_oid_key(rgw_obj& obj, rgw_bucket& bucket,
   prepend_bucket_marker(bucket, obj.key, key);
 }
 
+struct RGWObjVersionTracker {
+  obj_version read_version;
+  obj_version write_version;
+
+  obj_version *version_for_read() {
+    return &read_version;
+  }
+
+  obj_version *version_for_write() {
+    if (write_version.ver == 0)
+      return NULL;
+
+    return &write_version;
+  }
+
+  obj_version *version_for_check() {
+    if (read_version.ver == 0)
+      return NULL;
+
+    return &read_version;
+  }
+
+  void prepare_op_for_read(librados::ObjectReadOperation *op);
+  void prepare_op_for_write(librados::ObjectWriteOperation *op);
+
+  void apply_write() {
+    read_version = write_version;
+    write_version = obj_version();
+  }
+};
+
 struct RGWUsageBatch {
   map<utime_t, rgw_usage_log_entry> m;
 
@@ -182,7 +213,7 @@ struct RGWObjState {
   bool has_data;
   bufferlist data;
   bool prefetch_data;
-  obj_version objv;
+  RGWObjVersionTracker objv_tracker;
 
   map<string, bufferlist> attrset;
   RGWObjState() : is_atomic(false), has_attrs(0), exists(false),
@@ -466,7 +497,7 @@ class RGWRados
   Mutex bucket_id_lock;
   uint64_t max_bucket_id;
 
-  int get_obj_state(RGWRadosCtx *rctx, rgw_obj& obj, RGWObjState **state, obj_version *objv);
+  int get_obj_state(RGWRadosCtx *rctx, rgw_obj& obj, RGWObjState **state, RGWObjVersionTracker *objv_tracker);
   int append_atomic_test(RGWRadosCtx *rctx, rgw_obj& obj,
                          librados::ObjectOperation& op, RGWObjState **state);
   int prepare_atomic_for_write_impl(RGWRadosCtx *rctx, rgw_obj& obj,
@@ -641,11 +672,12 @@ public:
     const string *ptag;
     list<string> *remove_objs;
     bool modify_version;
-    obj_version *objv;
+    RGWObjVersionTracker *objv_tracker;
 
     PutObjMetaExtraParams() : mtime(NULL), rmattrs(NULL),
                      data(NULL), manifest(NULL), ptag(NULL),
-                     remove_objs(NULL), modify_version(false), objv(NULL) {}
+                     remove_objs(NULL), modify_version(false),
+                     objv_tracker(NULL) {}
   };
 
   /** Write/overwrite an object to the bucket storage. */
@@ -653,7 +685,7 @@ public:
               map<std::string, bufferlist>& attrs, RGWObjCategory category, int flags,
               map<std::string, bufferlist>* rmattrs, const bufferlist *data,
               RGWObjManifest *manifest, const string *ptag, list<string> *remove_objs,
-              bool modify_version, obj_version *objv);
+              bool modify_version, RGWObjVersionTracker *objv_tracker);
 
   virtual int put_obj_meta(void *ctx, rgw_obj& obj, uint64_t size,
               map<std::string, bufferlist>& attrs, RGWObjCategory category, int flags,
@@ -667,7 +699,7 @@ public:
                            RGWObjCategory category, int flags, PutObjMetaExtraParams& params) {
     return put_obj_meta_impl(ctx, obj, size, params.mtime, attrs, category, flags,
                         params.rmattrs, params.data, params.manifest, params.ptag, params.remove_objs,
-                        params.modify_version, params.objv);
+                        params.modify_version, params.objv_tracker);
   }
 
   virtual int put_obj_data(void *ctx, rgw_obj& obj, const char *data,
@@ -676,7 +708,7 @@ public:
                                off_t ofs, bool exclusive, void **handle);
   /* note that put_obj doesn't set category on an object, only use it for none user objects */
   int put_system_obj(void *ctx, rgw_obj& obj, const char *data, size_t len, bool exclusive,
-              time_t *mtime, map<std::string, bufferlist>& attrs, obj_version *objv) {
+              time_t *mtime, map<std::string, bufferlist>& attrs, RGWObjVersionTracker *objv_tracker) {
     bufferlist bl;
     bl.append(data, len);
     int flags = PUT_OBJ_CREATE;
@@ -687,7 +719,7 @@ public:
     ep.mtime = mtime;
     ep.data = &bl;
     ep.modify_version = true;
-    ep.objv = objv;
+    ep.objv_tracker = objv_tracker;
 
     int ret = put_obj_meta(ctx, obj, len, attrs, RGW_OBJ_CATEGORY_NONE, flags, ep);
     return ret;
@@ -828,11 +860,11 @@ public:
             const char *if_nomatch,
             uint64_t *total_size,
             uint64_t *obj_size,
-            obj_version *objv,
+            RGWObjVersionTracker *objv_tracker,
             void **handle,
             struct rgw_err *err);
 
-  virtual int get_obj(void *ctx, obj_version *objv, void **handle, rgw_obj& obj,
+  virtual int get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **handle, rgw_obj& obj,
                       bufferlist& bl, off_t ofs, off_t end);
 
   virtual void finish_get_obj(void **handle);
@@ -861,7 +893,7 @@ public:
 
   virtual int obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime,
                        uint64_t *epoch, map<string, bufferlist> *attrs, bufferlist *first_chunk,
-                       obj_version *objv);
+                       RGWObjVersionTracker *objv_tracker);
 
   virtual bool supports_omap() { return true; }
   virtual int omap_get_all(rgw_obj& obj, bufferlist& header, std::map<string, bufferlist>& m);
diff --git a/src/rgw/rgw_tools.cc b/src/rgw/rgw_tools.cc
index b36c182c2e4..473a8b245c2 100644
--- a/src/rgw/rgw_tools.cc
+++ b/src/rgw/rgw_tools.cc
@@ -15,7 +15,7 @@
 static map<string, string> ext_mime_map;
 
 int rgw_put_system_obj(RGWRados *rgwstore, rgw_bucket& bucket, string& oid, const char *data, size_t size, bool exclusive,
-                       obj_version *objv, map<string, bufferlist> *pattrs)
+                       RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs)
 {
   map<string,bufferlist> no_attrs;
   if (!pattrs)
@@ -23,18 +23,19 @@ int rgw_put_system_obj(RGWRados *rgwstore, rgw_bucket& bucket, string& oid, cons
 
   rgw_obj obj(bucket, oid);
 
-  int ret = rgwstore->put_system_obj(NULL, obj, data, size, exclusive, NULL, *pattrs, objv);
+  int ret = rgwstore->put_system_obj(NULL, obj, data, size, exclusive, NULL, *pattrs, objv_tracker);
 
   if (ret == -ENOENT) {
     ret = rgwstore->create_pool(bucket);
     if (ret >= 0)
-      ret = rgwstore->put_system_obj(NULL, obj, data, size, exclusive, NULL, *pattrs, objv);
+      ret = rgwstore->put_system_obj(NULL, obj, data, size, exclusive, NULL, *pattrs, objv_tracker);
   }
 
   return ret;
 }
 
-int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string& key, bufferlist& bl, obj_version *objv, map<string, bufferlist> *pattrs)
+int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string& key, bufferlist& bl,
+                       RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs)
 {
   int ret;
   struct rgw_err err;
@@ -42,13 +43,15 @@ int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string
   bufferlist::iterator iter;
   int request_len = READ_CHUNK_LEN;
   rgw_obj obj(bucket, key);
+
   do {
     ret = rgwstore->prepare_get_obj(ctx, obj, NULL, NULL, pattrs, NULL,
-                                  NULL, NULL, NULL, NULL, NULL, NULL, objv, &handle, &err);
+                                  NULL, NULL, NULL, NULL, NULL, NULL, objv_tracker, &handle, &err);
     if (ret < 0)
       return ret;
 
-    ret = rgwstore->get_obj(ctx, objv, &handle, obj, bl, 0, request_len - 1);
+    ret = rgwstore->get_obj(ctx, objv_tracker, &handle, obj, bl, 0, request_len - 1);
+#warning FIXME objv_tracker
     rgwstore->finish_get_obj(&handle);
     if (ret < 0)
       return ret;
diff --git a/src/rgw/rgw_tools.h b/src/rgw/rgw_tools.h
index 01b5b1bbcd2..5c79525ad8f 100644
--- a/src/rgw/rgw_tools.h
+++ b/src/rgw/rgw_tools.h
@@ -7,11 +7,14 @@
 #include "rgw_common.h"
 
 class RGWRados;
+class RGWObjVersionTracker;
 
 struct obj_version;
 
-int rgw_put_system_obj(RGWRados *rgwstore, rgw_bucket& bucket, string& oid, const char *data, size_t size, bool exclusive, obj_version *objv, map<string, bufferlist> *pattrs = NULL);
-int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string& key, bufferlist& bl, obj_version *objv, map<string, bufferlist> *pattrs = NULL);
+int rgw_put_system_obj(RGWRados *rgwstore, rgw_bucket& bucket, string& oid, const char *data, size_t size, bool exclusive,
+                       RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs = NULL);
+int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string& key, bufferlist& bl,
+                       RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs = NULL);
 
 int rgw_tools_init(CephContext *cct);
 void rgw_tools_cleanup();
diff --git a/src/rgw/rgw_user.cc b/src/rgw/rgw_user.cc
index f9d5de50a29..91329bf4538 100644
--- a/src/rgw/rgw_user.cc
+++ b/src/rgw/rgw_user.cc
@@ -17,6 +17,9 @@
 using namespace std;
 
 
+static RGWMetadataHandler *user_meta_handler = NULL;
+
+
 /**
  * Get the anonymous (ie, unauthenticated) user info.
  */
@@ -36,7 +39,7 @@ bool rgw_user_is_authenticated(RGWUserInfo& info)
  * Save the given user information to storage.
  * Returns: 0 on success, -ERR# on failure.
  */
-int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_info, obj_version *objv, bool exclusive)
+int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_info, RGWObjVersionTracker *objv_tracker, bool exclusive)
 {
   bufferlist bl;
   info.encode(bl);
@@ -82,14 +85,15 @@ int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_inf
   ::encode(ui, data_bl);
   ::encode(info, data_bl);
 
-  ret = rgw_put_system_obj(store, store->zone.user_uid_pool, info.user_id, data_bl.c_str(), data_bl.length(), exclusive, objv);
+  ret = store->meta_mgr->put_obj(user_meta_handler, info.user_id, data_bl, exclusive, objv_tracker);
   if (ret < 0)
     return ret;
 
   if (info.user_email.size()) {
     if (!old_info ||
         old_info->user_email.compare(info.user_email) != 0) { /* only if new index changed */
-      ret = rgw_put_system_obj(store, store->zone.user_email_pool, info.user_email, link_bl.c_str(), link_bl.length(), exclusive, objv);
+      ret = rgw_put_system_obj(store, store->zone.user_email_pool, info.user_email,
+                               link_bl.c_str(), link_bl.length(), exclusive, NULL);
       if (ret < 0)
         return ret;
     }
@@ -102,7 +106,9 @@ int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_inf
       if (old_info && old_info->access_keys.count(iter->first) != 0)
 	continue;
 
-      ret = rgw_put_system_obj(store, store->zone.user_keys_pool, k.id, link_bl.c_str(), link_bl.length(), exclusive, objv);
+      ret = rgw_put_system_obj(store, store->zone.user_keys_pool, k.id,
+                               link_bl.c_str(), link_bl.length(), exclusive,
+                               NULL);
       if (ret < 0)
         return ret;
     }
@@ -114,7 +120,9 @@ int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_inf
     if (old_info && old_info->swift_keys.count(siter->first) != 0)
       continue;
 
-    ret = rgw_put_system_obj(store, store->zone.user_swift_pool, k.id, link_bl.c_str(), link_bl.length(), exclusive, objv);
+    ret = rgw_put_system_obj(store, store->zone.user_swift_pool, k.id,
+                             link_bl.c_str(), link_bl.length(), exclusive,
+                             NULL);
     if (ret < 0)
       return ret;
   }
@@ -122,7 +130,7 @@ int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_inf
   return ret;
 }
 
-int rgw_get_user_info_from_index(RGWRados *store, string& key, rgw_bucket& bucket, RGWUserInfo& info)
+int rgw_get_user_info_from_index(RGWRados *store, string& key, rgw_bucket& bucket, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker)
 {
   bufferlist bl;
   RGWUID uid;
@@ -134,7 +142,7 @@ int rgw_get_user_info_from_index(RGWRados *store, string& key, rgw_bucket& bucke
   bufferlist::iterator iter = bl.begin();
   try {
     ::decode(uid, iter);
-    return rgw_get_user_info_by_uid(store, uid.user_id, info);
+    return rgw_get_user_info_by_uid(store, uid.user_id, info, objv_tracker);
   } catch (buffer::error& err) {
     ldout(store->ctx(), 0) << "ERROR: failed to decode user info, caught buffer::error" << dendl;
     return -EIO;
@@ -147,12 +155,12 @@ int rgw_get_user_info_from_index(RGWRados *store, string& key, rgw_bucket& bucke
  * Given a uid, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-static int rgw_read_uid_info(RGWRados *store, string& uid, RGWUserInfo& info, obj_version *objv)
+int rgw_get_user_info_by_uid(RGWRados *store, string& uid, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker)
 {
   bufferlist bl;
   RGWUID user_id;
 
-  int ret = rgw_get_system_obj(store, NULL, store->zone.user_uid_pool, uid, bl, objv);
+  int ret = rgw_get_system_obj(store, NULL, store->zone.user_uid_pool, uid, bl, objv_tracker);
   if (ret < 0)
     return ret;
 
@@ -174,36 +182,31 @@ static int rgw_read_uid_info(RGWRados *store, string& uid, RGWUserInfo& info, ob
   return 0;
 }
 
-int rgw_get_user_info_by_uid(RGWRados *store, string& uid, RGWUserInfo& info)
-{
-  return rgw_read_uid_info(store, uid, info, NULL);
-}
-
 /**
  * Given an email, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-int rgw_get_user_info_by_email(RGWRados *store, string& email, RGWUserInfo& info)
+int rgw_get_user_info_by_email(RGWRados *store, string& email, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker)
 {
-  return rgw_get_user_info_from_index(store, email, store->zone.user_email_pool, info);
+  return rgw_get_user_info_from_index(store, email, store->zone.user_email_pool, info, objv_tracker);
 }
 
 /**
  * Given an swift username, finds the user_info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-extern int rgw_get_user_info_by_swift(RGWRados *store, string& swift_name, RGWUserInfo& info)
+extern int rgw_get_user_info_by_swift(RGWRados *store, string& swift_name, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker)
 {
-  return rgw_get_user_info_from_index(store, swift_name, store->zone.user_swift_pool, info);
+  return rgw_get_user_info_from_index(store, swift_name, store->zone.user_swift_pool, info, objv_tracker);
 }
 
 /**
  * Given an access key, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-extern int rgw_get_user_info_by_access_key(RGWRados *store, string& access_key, RGWUserInfo& info)
+extern int rgw_get_user_info_by_access_key(RGWRados *store, string& access_key, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker)
 {
-  return rgw_get_user_info_from_index(store, access_key, store->zone.user_keys_pool, info);
+  return rgw_get_user_info_from_index(store, access_key, store->zone.user_keys_pool, info, objv_tracker);
 }
 
 static void get_buckets_obj(string& user_id, string& buckets_obj_id)
@@ -496,38 +499,44 @@ public:
   int get(RGWRados *store, string& entry, RGWMetadataObject **obj) {
     RGWUserInfo info;
 
-    obj_version objv;
+    RGWObjVersionTracker objv_tracker;
 
-    int ret = rgw_read_uid_info(store, entry, info, &objv);
+    int ret = rgw_get_user_info_by_uid(store, entry, info, &objv_tracker);
     if (ret < 0)
       return ret;
 
-    RGWUserMetadataObject *mdo = new RGWUserMetadataObject(info, objv);
+    RGWUserMetadataObject *mdo = new RGWUserMetadataObject(info, objv_tracker.read_version);
 
     *obj = mdo;
 
     return 0;
   }
 
-  int put(RGWRados *store, string& entry, obj_version& objv, JSONObj *obj) {
+  int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, JSONObj *obj) {
     RGWUserInfo info;
 
     decode_json_obj(info, obj);
 
     RGWUserInfo old_info;
-    obj_version old_objv;
-    int ret = rgw_read_uid_info(store, entry, old_info, &old_objv);
+    int ret = rgw_get_user_info_by_uid(store, entry, old_info, &objv_tracker);
     if (ret < 0)
       return ret;
 
 
-    ret = rgw_store_user_info(store, info, &old_info, &objv, false);
+    ret = rgw_store_user_info(store, info, &old_info, &objv_tracker, false);
     if (ret < 0)
       return ret;
 
     return 0;
   }
 
+  int put_obj(RGWRados *store, string& key, bufferlist& bl, bool exclusive,
+              RGWObjVersionTracker *objv_tracker, map<string, bufferlist> *pattrs) {
+    return rgw_put_system_obj(store, store->zone.user_uid_pool, key,
+                              bl.c_str(), bl.length(), exclusive,
+                              objv_tracker, pattrs);
+  }
+
   struct list_keys_info {
     RGWRados *store;
     RGWListRawObjsCtx ctx;
@@ -582,5 +591,6 @@ public:
 
 void rgw_user_init(RGWMetadataManager *mm)
 {
-  mm->register_handler(new RGWUserMetadataHandler);
+  user_meta_handler = new RGWUserMetadataHandler;
+  mm->register_handler(user_meta_handler);
 }
diff --git a/src/rgw/rgw_user.h b/src/rgw/rgw_user.h
index f4e0122e275..90ff18fb944 100644
--- a/src/rgw/rgw_user.h
+++ b/src/rgw/rgw_user.h
@@ -40,27 +40,27 @@ extern bool rgw_user_is_authenticated(RGWUserInfo& info);
  * Save the given user information to storage.
  * Returns: 0 on success, -ERR# on failure.
  */
-extern int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_info, obj_version *objv, bool exclusive);
+extern int rgw_store_user_info(RGWRados *store, RGWUserInfo& info, RGWUserInfo *old_info, RGWObjVersionTracker *objv_tracker, bool exclusive);
 /**
  * Given an email, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-extern int rgw_get_user_info_by_uid(RGWRados *store, string& user_id, RGWUserInfo& info);
+extern int rgw_get_user_info_by_uid(RGWRados *store, string& user_id, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker = NULL);
 /**
  * Given an swift username, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-extern int rgw_get_user_info_by_email(RGWRados *store, string& email, RGWUserInfo& info);
+extern int rgw_get_user_info_by_email(RGWRados *store, string& email, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker = NULL);
 /**
  * Given an swift username, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-extern int rgw_get_user_info_by_swift(RGWRados *store, string& swift_name, RGWUserInfo& info);
+extern int rgw_get_user_info_by_swift(RGWRados *store, string& swift_name, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker = NULL);
 /**
  * Given an access key, finds the user info associated with it.
  * returns: 0 on success, -ERR# on failure (including nonexistence)
  */
-extern int rgw_get_user_info_by_access_key(RGWRados *store, string& access_key, RGWUserInfo& info);
+extern int rgw_get_user_info_by_access_key(RGWRados *store, string& access_key, RGWUserInfo& info, RGWObjVersionTracker *objv_tracker = NULL);
 /**
  * Given an RGWUserInfo, deletes the user and its bucket ACLs.
  */