diff --git a/src/gen-vdl/v.io/syncbase/v23/services/syncbase/nosql/index.js b/src/gen-vdl/v.io/syncbase/v23/services/syncbase/nosql/index.js
index 1e5f77d..168e415 100644
--- a/src/gen-vdl/v.io/syncbase/v23/services/syncbase/nosql/index.js
+++ b/src/gen-vdl/v.io/syncbase/v23/services/syncbase/nosql/index.js
@@ -29,11 +29,11 @@
 var _type6 = new vdl.Type();
 var _type7 = new vdl.Type();
 var _typeBatchOptions = new vdl.Type();
+var _typeBlobFetchState = new vdl.Type();
+var _typeBlobFetchStatus = new vdl.Type();
 var _typeBlobRef = new vdl.Type();
 var _typeCrPolicy = new vdl.Type();
 var _typeCrRule = new vdl.Type();
-var _typeFetchState = new vdl.Type();
-var _typeFetchStatus = new vdl.Type();
 var _typeKeyValue = new vdl.Type();
 var _typePrefixPermissions = new vdl.Type();
 var _typeResolverType = new vdl.Type();
@@ -66,6 +66,12 @@
 _typeBatchOptions.kind = vdl.kind.STRUCT;
 _typeBatchOptions.name = "v.io/syncbase/v23/services/syncbase/nosql.BatchOptions";
 _typeBatchOptions.fields = [{name: "Hint", type: vdl.types.STRING}, {name: "ReadOnly", type: vdl.types.BOOL}];
+_typeBlobFetchState.kind = vdl.kind.ENUM;
+_typeBlobFetchState.name = "v.io/syncbase/v23/services/syncbase/nosql.BlobFetchState";
+_typeBlobFetchState.labels = ["Pending", "Locating", "Fetching", "Done"];
+_typeBlobFetchStatus.kind = vdl.kind.STRUCT;
+_typeBlobFetchStatus.name = "v.io/syncbase/v23/services/syncbase/nosql.BlobFetchStatus";
+_typeBlobFetchStatus.fields = [{name: "State", type: _typeBlobFetchState}, {name: "Received", type: vdl.types.INT64}, {name: "Total", type: vdl.types.INT64}];
 _typeBlobRef.kind = vdl.kind.STRING;
 _typeBlobRef.name = "v.io/syncbase/v23/services/syncbase/nosql.BlobRef";
 _typeCrPolicy.kind = vdl.kind.STRUCT;
@@ -74,12 +80,6 @@
 _typeCrRule.kind = vdl.kind.STRUCT;
 _typeCrRule.name = "v.io/syncbase/v23/services/syncbase/nosql.CrRule";
 _typeCrRule.fields = [{name: "TableName", type: vdl.types.STRING}, {name: "KeyPrefix", type: vdl.types.STRING}, {name: "Type", type: vdl.types.STRING}, {name: "Resolver", type: _typeResolverType}];
-_typeFetchState.kind = vdl.kind.ENUM;
-_typeFetchState.name = "v.io/syncbase/v23/services/syncbase/nosql.FetchState";
-_typeFetchState.labels = ["Pending", "Locating", "Fetching", "Done"];
-_typeFetchStatus.kind = vdl.kind.STRUCT;
-_typeFetchStatus.name = "v.io/syncbase/v23/services/syncbase/nosql.FetchStatus";
-_typeFetchStatus.fields = [{name: "State", type: _typeFetchState}, {name: "Received", type: vdl.types.UINT64}, {name: "Total", type: vdl.types.UINT64}];
 _typeKeyValue.kind = vdl.kind.STRUCT;
 _typeKeyValue.name = "v.io/syncbase/v23/services/syncbase/nosql.KeyValue";
 _typeKeyValue.fields = [{name: "Key", type: vdl.types.STRING}, {name: "Value", type: _type3}];
@@ -109,11 +109,11 @@
 _type6.freeze();
 _type7.freeze();
 _typeBatchOptions.freeze();
+_typeBlobFetchState.freeze();
+_typeBlobFetchStatus.freeze();
 _typeBlobRef.freeze();
 _typeCrPolicy.freeze();
 _typeCrRule.freeze();
-_typeFetchState.freeze();
-_typeFetchStatus.freeze();
 _typeKeyValue.freeze();
 _typePrefixPermissions.freeze();
 _typeResolverType.freeze();
@@ -122,16 +122,16 @@
 _typeSyncGroupMemberInfo.freeze();
 _typeSyncGroupSpec.freeze();
 module.exports.BatchOptions = (vdl.registry.lookupOrCreateConstructor(_typeBatchOptions));
+module.exports.BlobFetchState = {
+  PENDING: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeBlobFetchState))('Pending', true), _typeBlobFetchState),
+  LOCATING: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeBlobFetchState))('Locating', true), _typeBlobFetchState),
+  FETCHING: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeBlobFetchState))('Fetching', true), _typeBlobFetchState),
+  DONE: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeBlobFetchState))('Done', true), _typeBlobFetchState),
+};
+module.exports.BlobFetchStatus = (vdl.registry.lookupOrCreateConstructor(_typeBlobFetchStatus));
 module.exports.BlobRef = (vdl.registry.lookupOrCreateConstructor(_typeBlobRef));
 module.exports.CrPolicy = (vdl.registry.lookupOrCreateConstructor(_typeCrPolicy));
 module.exports.CrRule = (vdl.registry.lookupOrCreateConstructor(_typeCrRule));
-module.exports.FetchState = {
-  PENDING: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeFetchState))('Pending', true), _typeFetchState),
-  LOCATING: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeFetchState))('Locating', true), _typeFetchState),
-  FETCHING: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeFetchState))('Fetching', true), _typeFetchState),
-  DONE: canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeFetchState))('Done', true), _typeFetchState),
-};
-module.exports.FetchStatus = (vdl.registry.lookupOrCreateConstructor(_typeFetchStatus));
 module.exports.KeyValue = (vdl.registry.lookupOrCreateConstructor(_typeKeyValue));
 module.exports.PrefixPermissions = (vdl.registry.lookupOrCreateConstructor(_typePrefixPermissions));
 module.exports.ResolverType = {
@@ -185,6 +185,12 @@
 ]);
 
 
+module.exports.BlobNotCommittedError = makeError('v.io/syncbase/v23/services/syncbase/nosql.BlobNotCommitted', actions.NO_RETRY, {
+  'en': '{1:}{2:} blob is not yet committed',
+}, [
+]);
+
+
 
 
 // Services:
@@ -210,7 +216,7 @@
 DatabaseWatcher.prototype._serviceDescription = {
   name: 'DatabaseWatcher',
   pkgPath: 'v.io/syncbase/v23/services/syncbase/nosql',
-  doc: "// Watch allows a client to watch for updates in the database. For each watched\n// request, the client will receive a reliable stream of watch events without\n// re-ordering. See watch.GlobWatcher for a detailed explanation of the\n// behavior.\n//\n// The watching is done by starting a streaming RPC. The argument to the RPC\n// contains the ResumeMarker that points to a particular place in the database\n// event log. The result stream consists of a never-ending sequence of Change\n// messages (until the call fails or is canceled). Each Change contains the\n// Name field in the form \"<tableName>/<rowKey>\" and the Value field of the\n// StoreChange type. If the client has no access to a row specified in a change,\n// that change is excluded from the result stream.\n//\n// The DatabaseWatcher is designed to be used in the following way:\n// 1) begin a read-only batch\n// 2) read all information your app needs\n// 3) read the ResumeMarker\n// 4) abort the batch\n// 5) start watching changes to the data using the ResumeMarker\n// In this configuration the client doesn't miss any changes.",
+  doc: "// DatabaseWatcher allows a client to watch for updates in the database.\n// For each watched request, the client will receive a reliable stream of watch\n// events without re-ordering. See watch.GlobWatcher for a detailed explanation\n// of the behavior.\n// TODO(rogulenko): Currently the only supported watch patterns are\n// 'table/row*'. Consider changing that.\n//\n// The watching is done by starting a streaming RPC. The argument to the RPC\n// contains the ResumeMarker that points to a particular place in the database\n// event log. The result stream consists of a never-ending sequence of Change\n// messages (until the call fails or is canceled). Each Change contains the\n// Name field in the form \"<tableName>/<rowKey>\" and the Value field of the\n// StoreChange type. If the client has no access to a row specified in a change,\n// that change is excluded from the result stream.\n//\n// The DatabaseWatcher is designed to be used in the following way:\n// 1) begin a read-only batch\n// 2) read all information your app needs\n// 3) read the ResumeMarker\n// 4) abort the batch\n// 5) start watching changes to the data using the ResumeMarker\n// In this configuration the client doesn't miss any changes.",
   embeds: [{
       name: 'GlobWatcher',
       pkgPath: 'v.io/v23/services/watch',
@@ -714,7 +720,7 @@
     outStream: {
       name: '',
       doc: '',
-      type: _typeFetchStatus
+      type: _typeBlobFetchStatus
     },
     tags: [canonicalize.reduce(new access.Tag("Read", true), new access.Tag()._type), ]
   },
@@ -857,6 +863,11 @@
 };
     
       
+Database.prototype.exec = function(ctx, serverCall, schemaVersion, query) {
+  throw new Error('Method Exec not implemented');
+};
+    
+      
 Database.prototype.beginBatch = function(ctx, serverCall, schemaVersion, bo) {
   throw new Error('Method BeginBatch not implemented');
 };
@@ -867,11 +878,6 @@
 };
     
       
-Database.prototype.exec = function(ctx, serverCall, schemaVersion, query) {
-  throw new Error('Method Exec not implemented');
-};
-    
-      
 Database.prototype.abort = function(ctx, serverCall, schemaVersion) {
   throw new Error('Method Abort not implemented');
 };
@@ -1015,7 +1021,7 @@
     {
       name: 'DatabaseWatcher',
       pkgPath: 'v.io/syncbase/v23/services/syncbase/nosql',
-      doc: "// Watch allows a client to watch for updates in the database. For each watched\n// request, the client will receive a reliable stream of watch events without\n// re-ordering. See watch.GlobWatcher for a detailed explanation of the\n// behavior.\n//\n// The watching is done by starting a streaming RPC. The argument to the RPC\n// contains the ResumeMarker that points to a particular place in the database\n// event log. The result stream consists of a never-ending sequence of Change\n// messages (until the call fails or is canceled). Each Change contains the\n// Name field in the form \"<tableName>/<rowKey>\" and the Value field of the\n// StoreChange type. If the client has no access to a row specified in a change,\n// that change is excluded from the result stream.\n//\n// The DatabaseWatcher is designed to be used in the following way:\n// 1) begin a read-only batch\n// 2) read all information your app needs\n// 3) read the ResumeMarker\n// 4) abort the batch\n// 5) start watching changes to the data using the ResumeMarker\n// In this configuration the client doesn't miss any changes."
+      doc: "// DatabaseWatcher allows a client to watch for updates in the database.\n// For each watched request, the client will receive a reliable stream of watch\n// events without re-ordering. See watch.GlobWatcher for a detailed explanation\n// of the behavior.\n// TODO(rogulenko): Currently the only supported watch patterns are\n// 'table/row*'. Consider changing that.\n//\n// The watching is done by starting a streaming RPC. The argument to the RPC\n// contains the ResumeMarker that points to a particular place in the database\n// event log. The result stream consists of a never-ending sequence of Change\n// messages (until the call fails or is canceled). Each Change contains the\n// Name field in the form \"<tableName>/<rowKey>\" and the Value field of the\n// StoreChange type. If the client has no access to a row specified in a change,\n// that change is excluded from the result stream.\n//\n// The DatabaseWatcher is designed to be used in the following way:\n// 1) begin a read-only batch\n// 2) read all information your app needs\n// 3) read the ResumeMarker\n// 4) abort the batch\n// 5) start watching changes to the data using the ResumeMarker\n// In this configuration the client doesn't miss any changes."
     },
     {
       name: 'SyncGroupManager',
@@ -1095,6 +1101,31 @@
     
       
     {
+    name: 'Exec',
+    doc: "// Exec executes a syncQL query and returns all results as specified by in the\n// query's select clause. Concurrency semantics are documented in model.go.",
+    inArgs: [{
+      name: 'schemaVersion',
+      doc: "",
+      type: vdl.types.INT32
+    },
+    {
+      name: 'query',
+      doc: "",
+      type: vdl.types.STRING
+    },
+    ],
+    outArgs: [],
+    inStream: null,
+    outStream: {
+      name: '',
+      doc: '',
+      type: _type6
+    },
+    tags: [canonicalize.reduce(new access.Tag("Read", true), new access.Tag()._type), ]
+  },
+    
+      
+    {
     name: 'BeginBatch',
     doc: "// BeginBatch creates a new batch. It returns an App-relative name for a\n// Database handle bound to this batch. If this Database is already bound to a\n// batch, BeginBatch() will fail with ErrBoundToBatch. Concurrency semantics\n// are documented in model.go.\n// TODO(sadovsky): make BatchOptions optional",
     inArgs: [{
@@ -1137,31 +1168,6 @@
     
       
     {
-    name: 'Exec',
-    doc: "// Exec executes a syncQL query and returns all results as specified by in the\n// query's select clause. Concurrency semantics are documented in model.go.",
-    inArgs: [{
-      name: 'schemaVersion',
-      doc: "",
-      type: vdl.types.INT32
-    },
-    {
-      name: 'query',
-      doc: "",
-      type: vdl.types.STRING
-    },
-    ],
-    outArgs: [],
-    inStream: null,
-    outStream: {
-      name: '',
-      doc: '',
-      type: _type6
-    },
-    tags: [canonicalize.reduce(new access.Tag("Read", true), new access.Tag()._type), ]
-  },
-    
-      
-    {
     name: 'Abort',
     doc: "// Abort notifies the server that any pending changes can be discarded.\n// It is not strictly required, but it may allow the server to release locks\n// or other resources sooner than if it was not called.\n// If this Database is not bound to a batch, Abort() will fail with\n// ErrNotBoundToBatch.",
     inArgs: [{
@@ -1582,7 +1588,7 @@
     outStream: {
       name: '',
       doc: '',
-      type: _typeFetchStatus
+      type: _typeBlobFetchStatus
     },
     tags: [canonicalize.reduce(new access.Tag("Read", true), new access.Tag()._type), ]
   },
@@ -1791,7 +1797,7 @@
       
     {
     name: 'DeleteRowRange',
-    doc: "// Delete deletes all rows in the given half-open range [start, limit). If\n// limit is \"\", all rows with keys >= start are included.",
+    doc: "// Delete deletes all rows in the given half-open range [start, limit). If\n// limit is \"\", all rows with keys >= start are included.\n// TODO(sadovsky): Delete prefix perms fully covered by the row range?",
     inArgs: [{
       name: 'schemaVersion',
       doc: "",
diff --git a/src/gen-vdl/v.io/v23/security/index.js b/src/gen-vdl/v.io/v23/security/index.js
index 5221584..de0e2eb 100644
--- a/src/gen-vdl/v.io/v23/security/index.js
+++ b/src/gen-vdl/v.io/v23/security/index.js
@@ -273,17 +273,17 @@
 
   module.exports.SHA512Hash = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(_typeHash))("SHA512", true), _typeHash);
 
-  module.exports.SignatureForMessageSigning = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("S", true), vdl.types.STRING);
+  module.exports.SignatureForMessageSigning = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("S1", true), vdl.types.STRING);
 
-  module.exports.SignatureForBlessingCertificates = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("B", true), vdl.types.STRING);
+  module.exports.SignatureForBlessingCertificates = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("B1", true), vdl.types.STRING);
 
-  module.exports.SignatureForDischarge = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("D", true), vdl.types.STRING);
+  module.exports.SignatureForDischarge = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("D1", true), vdl.types.STRING);
 
-  module.exports.SignatureForMessageSigningV1 = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("S1", true), vdl.types.STRING);
+  module.exports.SignatureForMessageSigningV0 = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("S", true), vdl.types.STRING);
 
-  module.exports.SignatureForBlessingCertificatesV1 = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("B1", true), vdl.types.STRING);
+  module.exports.SignatureForBlessingCertificatesV0 = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("B", true), vdl.types.STRING);
 
-  module.exports.SignatureForDischargeV1 = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("D1", true), vdl.types.STRING);
+  module.exports.SignatureForDischargeV0 = canonicalize.reduce(new (vdl.registry.lookupOrCreateConstructor(vdl.types.STRING))("D", true), vdl.types.STRING);
 
 
 
diff --git a/src/nosql/database.js b/src/nosql/database.js
index 58c0993..febf66f 100644
--- a/src/nosql/database.js
+++ b/src/nosql/database.js
@@ -10,11 +10,14 @@
 // vanadium.vdl object.
 var unwrap = require('vanadium/src/vdl/type-util').unwrap;
 
-var BatchDatabase = require('./batch-database');
 var nosqlVdl = require('../gen-vdl/v.io/syncbase/v23/services/syncbase/nosql');
+var watchVdl = require('../gen-vdl/v.io/v23/services/watch');
+
+var BatchDatabase = require('./batch-database');
 var SyncGroup = require('./syncgroup');
 var Table = require('./table');
 var util = require('../util');
+var watch = require('./watch');
 
 /**
  * Database represents a collection of Tables. Batches, queries, sync, watch,
@@ -214,6 +217,80 @@
 };
 
 /**
+ * Watches for updates to the database. For each watch request, the client will
+ * receive a reliable stream of watch events without re-ordering.
+ *
+ * This method is designed to be used in the following way:
+ * 1) begin a read-only batch
+ * 2) read all information your app needs
+ * 3) read the ResumeMarker
+ * 4) abort the batch
+ * 5) start watching for changes to the data using the ResumeMarker
+ *
+ * In this configuration the client doesn't miss any changes.
+ *
+ * @param {module:vanadium.context.Context} ctx Vanadium context.
+ * @param {string} table Name of table to watch.
+ * @param {string} prefix Prefix of keys to watch.
+ * @param {module:syncbase.nosql.watch.ResumeMarker} resumeMarker ResumeMarker
+ * to resume watching from.
+ * @param {function} [cb] Optional callback that will be called after watch RPC
+ * finishes.
+ * @returns {stream} Stream of WatchChange objects.
+ */
+Database.prototype.watch = function(ctx, tableName, prefix, resumeMarker, cb) {
+  var globReq = new watchVdl.GlobRequest({
+    pattern: vanadium.naming.join(tableName, prefix + '*'),
+    resumeMarker: resumeMarker
+  });
+
+  var watchChangeEncoder = through2({
+    objectMode: true
+  }, function(change, enc, cb) {
+    var changeType;
+    switch (change.state) {
+      case watchVdl.Exists.val:
+        changeType = 'put';
+        break;
+      case watchVdl.DoesNotExist.val:
+        changeType = 'delete';
+        break;
+      default:
+        return cb(new Error('invalid change state ' + change.state));
+    }
+
+    var wc = new watch.WatchChange({
+      tableName: vanadium.naming.stripBasename(change.name),
+      rowName: vanadium.naming.basename(change.name),
+      changeType: changeType,
+      valueBytes: changeType === 'put' ? change.value.value : null,
+      resumeMarker: change.resumeMarker,
+      fromSync: change.value.fromSync,
+      continued: change.continued
+    });
+    return cb(null, wc);
+  });
+
+  var stream = this._wire(ctx).watchGlob(ctx, globReq, cb).stream;
+
+  var watchChangeStream = stream.pipe(watchChangeEncoder);
+  stream.on('error', function(err) {
+    watchChangeStream.emit('error', err);
+  });
+
+  return watchChangeStream;
+};
+
+/**
+ * Gets the ResumeMarker that points to the current end of the event log.
+ * @param {module:vanadium.context.Context} ctx Vanadium context.
+ * @param {function} cb Callback.
+ */
+Database.prototype.getResumeMarker = function(ctx, cb) {
+  this._wire(ctx).getResumeMarker(ctx, cb);
+};
+
+/**
  * Replaces the current Permissions for the Database.
  * @param {module:vanadium.context.Context} ctx Vanadium context.
  * @param {module:vanadium.security.access.Permissions} perms Permissions for
diff --git a/src/nosql/index.js b/src/nosql/index.js
index c589058..5bbab69 100644
--- a/src/nosql/index.js
+++ b/src/nosql/index.js
@@ -5,6 +5,7 @@
 var rowrange = require('./rowrange');
 var runInBatch = require('./batch');
 var vdl = require('../gen-vdl/v.io/syncbase/v23/services/syncbase/nosql');
+var watch = require('./watch');
 
 /**
  * @summary
@@ -16,8 +17,10 @@
 module.exports = {
   BatchOptions: vdl.BatchOptions,
   ReadOnlyBatchError: vdl.ReadOnlyBatchError,
+  ResumeMarker: watch.ResumeMarker,
   rowrange: rowrange,
   runInBatch: runInBatch,
   SyncGroupMemberInfo: vdl.SyncGroupMemberInfo,
-  SyncGroupSpec: vdl.SyncGroupSpec
+  SyncGroupSpec: vdl.SyncGroupSpec,
+  WatchChange: watch.WatchChange
 };
diff --git a/src/nosql/watch.js b/src/nosql/watch.js
new file mode 100644
index 0000000..dba87e2
--- /dev/null
+++ b/src/nosql/watch.js
@@ -0,0 +1,107 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+var vom = require('vanadium').vom;
+
+var watchVdl = require('../gen-vdl/v.io/v23/services/watch');
+
+module.exports = {
+  ResumeMarker: watchVdl.ResumeMarker,
+  WatchChange: WatchChange
+};
+
+/**
+ * WatchChange represents the new value for a watched entity.
+ * @constructor
+ */
+function WatchChange(opts) {
+  /**
+   * @property tableName
+   * The name of the table that contains the changed row
+   */
+  Object.defineProperty(this, 'tableName', {
+    enumerable: true,
+    value: opts.tableName,
+    writable: false
+  });
+
+  /**
+   * @property rowName
+   * Name of the changed row.
+   */
+  Object.defineProperty(this, 'rowName', {
+    enumerable: true,
+    value: opts.rowName,
+    writable: false
+  });
+
+  /**
+   * @property changeType
+   * Describes the type of the change. If the changeType equals 'put', then the
+   * row exists in the table and the value contains the new value for this row.
+   * If the state equals 'delete', then the row was removed from the table.
+   */
+  Object.defineProperty(this, 'changeType', {
+    enumerable: true,
+    value: opts.changeType,
+    writable: false
+  });
+
+  /**
+   * @property valueBytes
+   * The new VOM-encoded value for the row if the changeType is 'put' or nil
+   * otherwise.
+   */
+  Object.defineProperty(this, 'valueBytes', {
+    enumerable: true,
+    value: opts.valueBytes,
+    writable: false
+  });
+
+  /**
+   * @property resumeMarker
+   * Provides a compact representation of all the messages that have been
+   * received by the caller for the given watch call.  This marker can be
+   * provided in the request message to allow the caller to resume the stream
+   * watching at a specific point without fetching the initial state.
+   */
+  Object.defineProperty(this, 'resumeMarker', {
+    enumerable: true,
+    value: opts.resumeMarker,
+    writable: false
+  });
+
+  /**
+   * @property fromSync
+   * Indicates whether the change came from sync. If fromSync is false, then
+   * the change originated from the local device.
+   */
+  Object.defineProperty(this, 'fromSync', {
+    enumerable: true,
+    value: opts.fromSync || false,
+    writable: false
+  });
+
+  /**
+   * @property continued
+   * If true, this WatchChange is followed by more WatchChanges that are in the
+   * same batch as this WatchChange
+   */
+  Object.defineProperty(this, 'continued', {
+    enumerable: true,
+    value: opts.continued || false,
+    writable: false
+  });
+}
+
+/**
+ * Decodes the new value of the watched element.
+ */
+WatchChange.prototype.getValue = function(cb) {
+  if (this.changeType === 'delete') {
+    return cb(new Error('invalid change type'));
+  }
+
+  vom.decode(this.valueBytes, false, null, cb);
+};
diff --git a/test/integration/test-watch.js b/test/integration/test-watch.js
new file mode 100644
index 0000000..fca1cec
--- /dev/null
+++ b/test/integration/test-watch.js
@@ -0,0 +1,132 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+var async = require('async');
+var test = require('prova');
+var vom = require('vanadium').vom;
+
+var syncbase = require('../..');
+var WatchChange = syncbase.nosql.WatchChange;
+
+var testUtil = require('./util');
+var setupTable = testUtil.setupTable;
+var uniqueName = testUtil.uniqueName;
+
+// Tests the basic client watch functionality (no perms or batches).  First
+// does some puts and deletes, fetching a ResumeMarker after each operation.
+// Then calls 'watch' with different prefixes and ResumeMarkers and verifies
+// that the resulting stream contains the correct changes.
+test('basic client watch', function(t) {
+  setupTable(t, function(err, o) {
+    if (err) {
+      return t.end(err);
+    }
+
+    var ctx = o.ctx;
+    var db = o.database;
+    var table = o.table;
+
+    var row1Prefix = 'row-abc';
+    var row1 = table.row(uniqueName(row1Prefix));
+    var value1 = uniqueName('value');
+
+    var row2Prefix = 'row-a';
+    var row2 = table.row(uniqueName(row2Prefix));
+    var value2 = uniqueName('value');
+
+    var resumeMarkers = [];
+
+    function getAndAppendResumeMarker(cb) {
+      db.getResumeMarker(ctx, function(err, rm) {
+        if (err) {
+          return cb(err);
+        }
+        resumeMarkers.push(rm);
+        cb(null);
+      });
+    }
+
+    // Generate the data and resume markers.
+    async.waterfall([
+      // Initial state.
+      getAndAppendResumeMarker,
+
+      // Put to row1.
+      row1.put.bind(row1, ctx, value1),
+      getAndAppendResumeMarker,
+
+      // Delete row1.
+      row1.delete.bind(row1, ctx),
+      getAndAppendResumeMarker,
+
+      // Put to row2.
+      row2.put.bind(row2, ctx, value2),
+      getAndAppendResumeMarker
+    ], assertCorrectChanges);
+
+    function assertCorrectChanges(err) {
+      if (err) {
+        t.error(err);
+        return o.teardown(t.end);
+      }
+
+      var allExpectedChanges = [new WatchChange({
+        tableName: table.name,
+        rowName: row1.key,
+        changeType: 'put',
+        valueBytes: vom.encode(value1),
+        resumeMarker: resumeMarkers[1]
+      }), new WatchChange({
+        tableName: table.name,
+        rowName: row1.key,
+        changeType: 'delete',
+        valueBytes: null,
+        resumeMarker: resumeMarkers[2]
+      }), new WatchChange({
+        tableName: table.name,
+        rowName: row2.key,
+        changeType: 'put',
+        valueBytes: vom.encode(value2),
+        resumeMarker: resumeMarkers[3]
+      })];
+
+      async.series([
+        assertWatch.bind(null, t, ctx, db, table.name, row2Prefix,
+                         resumeMarkers[0], allExpectedChanges),
+        assertWatch.bind(null, t, ctx, db, table.name, row2Prefix,
+                         resumeMarkers[1], allExpectedChanges.slice(1)),
+        assertWatch.bind(null, t, ctx, db, table.name, row2Prefix,
+                         resumeMarkers[2], allExpectedChanges.slice(2)),
+
+        assertWatch.bind(null, t, ctx, db, table.name, row1Prefix,
+                         resumeMarkers[0], allExpectedChanges.slice(0,2)),
+        assertWatch.bind(null, t, ctx, db, table.name, row1Prefix,
+                         resumeMarkers[1], allExpectedChanges.slice(1,2)),
+      ], function(err) {
+        t.error(err);
+        o.teardown(t.end);
+      });
+    }
+  });
+});
+
+function assertWatch(t, ctx, db, tableName, rowPrefix, resumeMarker,
+                     expectedWatchChanges, cb) {
+  var cctx = ctx.withCancel();
+  var stream = db.watch(ctx, tableName, rowPrefix, resumeMarker);
+
+  async.timesSeries(expectedWatchChanges.length, function(i, next) {
+    stream.once('data', function(gotWatchChange) {
+      t.deepEqual(gotWatchChange, expectedWatchChanges[i]);
+
+      next(null);
+    });
+  }, function(err) {
+    cctx.finish();
+    if (err) {
+      return cb(err);
+    }
+    cb(null);
+  });
+}