diff --git a/src/ledger/bin/BUILD.gn b/src/ledger/bin/BUILD.gn
index 36d1ba0..042abc8 100644
--- a/src/ledger/bin/BUILD.gn
+++ b/src/ledger/bin/BUILD.gn
@@ -71,7 +71,10 @@
 }
 
 fuzz_package("ledger_fuzzers") {
-  targets = [ "//src/ledger/bin/storage/impl/btree:encoding_fuzzer" ]
+  targets = [
+    "//src/ledger/bin/p2p_sync/impl:p2p_sync_fuzzer",
+    "//src/ledger/bin/storage/impl/btree:encoding_fuzzer",
+  ]
   sanitizers = [
     "asan",
     "ubsan",
diff --git a/src/ledger/bin/p2p_sync/impl/BUILD.gn b/src/ledger/bin/p2p_sync/impl/BUILD.gn
index cec55e3..a6aeeb2 100644
--- a/src/ledger/bin/p2p_sync/impl/BUILD.gn
+++ b/src/ledger/bin/p2p_sync/impl/BUILD.gn
@@ -2,6 +2,7 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+import("//build/fuzzing/fuzzer.gni")
 import("//third_party/flatbuffers/flatbuffer.gni")
 
 visibility = [ "//src/ledger/*" ]
@@ -11,6 +12,8 @@
     "commit_batch.cc",
     "commit_batch.h",
     "device_mesh.h",
+    "encoding.cc",
+    "encoding.h",
     "flatbuffer_message_factory.cc",
     "flatbuffer_message_factory.h",
     "ledger_communicator_impl.cc",
@@ -51,12 +54,14 @@
   testonly = true
 
   sources = [
+    "encoding_unittest.cc",
     "page_communicator_impl_unittest.cc",
     "user_communicator_impl_unittest.cc",
   ]
 
   deps = [
     ":impl",
+    ":message",
     "//garnet/public/lib/fsl",
     "//sdk/lib/fidl/cpp",
     "//src/ledger/bin/p2p_provider/impl",
@@ -73,3 +78,14 @@
 
   configs += [ "//src/ledger:ledger_config" ]
 }
+
+fuzz_target("p2p_sync_fuzzer") {
+  sources = [
+    "user_communicator_impl_fuzztest.cc",
+  ]
+  deps = [
+    ":impl",
+    "//src/ledger/bin/p2p_provider/public",
+    "//src/ledger/bin/storage/testing",
+  ]
+}
diff --git a/src/ledger/bin/p2p_sync/impl/encoding.cc b/src/ledger/bin/p2p_sync/impl/encoding.cc
new file mode 100644
index 0000000..2f4a241
--- /dev/null
+++ b/src/ledger/bin/p2p_sync/impl/encoding.cc
@@ -0,0 +1,193 @@
+// Copyright 2019 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "src/ledger/bin/p2p_sync/impl/encoding.h"
+
+#include "flatbuffers/flatbuffers.h"
+#include "src/ledger/bin/p2p_sync/impl/message_generated.h"
+#include "src/lib/fxl/strings/string_view.h"
+
+namespace p2p_sync {
+namespace {
+
+bool IsValidCommitRequest(flatbuffers::Verifier& verifier,
+                          const CommitRequest* request) {
+  return request && request->Verify(verifier) && request->commit_ids();
+}
+
+bool IsValidObjectRequest(flatbuffers::Verifier& verifier,
+                          const ObjectRequest* request) {
+  return request && request->Verify(verifier) && request->object_ids();
+}
+
+bool IsValidWatchStartRequest(flatbuffers::Verifier& verifier,
+                              const WatchStartRequest* request) {
+  return request && request->Verify(verifier);
+}
+
+bool IsValidWatchStopRequest(flatbuffers::Verifier& verifier,
+                             const WatchStopRequest* request) {
+  return request && request->Verify(verifier);
+}
+
+bool IsValidRequest(flatbuffers::Verifier& verifier, const Request* request) {
+  if (!request || !request->Verify(verifier)) {
+    return false;
+  }
+
+  const NamespacePageId* namespace_page_id = request->namespace_page();
+  if (!namespace_page_id || !namespace_page_id->Verify(verifier)) {
+    return false;
+  }
+
+  if (!namespace_page_id->namespace_id() || !namespace_page_id->page_id()) {
+    return false;
+  }
+
+  switch (request->request_type()) {
+    case RequestMessage_NONE:
+      return false;
+    case RequestMessage_CommitRequest:
+      return IsValidCommitRequest(
+          verifier, static_cast<const CommitRequest*>(request->request()));
+    case RequestMessage_ObjectRequest:
+      return IsValidObjectRequest(
+          verifier, static_cast<const ObjectRequest*>(request->request()));
+    case RequestMessage_WatchStartRequest:
+      return IsValidWatchStartRequest(
+          verifier, static_cast<const WatchStartRequest*>(request->request()));
+    case RequestMessage_WatchStopRequest:
+      return IsValidWatchStopRequest(
+          verifier, static_cast<const WatchStopRequest*>(request->request()));
+  }
+}
+
+bool IsValidCommitResponse(flatbuffers::Verifier& verifier,
+                           const CommitResponse* response) {
+  if (!response || !response->Verify(verifier)) {
+    return false;
+  }
+
+  if (!response->commits()) {
+    return false;
+  }
+
+  for (const Commit* commit : *response->commits()) {
+    if (!commit || !commit->Verify(verifier)) {
+      return false;
+    }
+
+    const CommitId* id = commit->id();
+    if (!id || !id->Verify(verifier) || !id->id()) {
+      return false;
+    }
+
+    const Data* data = commit->commit();
+    if (!data || !data->Verify(verifier) || !data->bytes()) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+bool IsValidObjectResponse(flatbuffers::Verifier& verifier,
+                           const ObjectResponse* response) {
+  if (!response || !response->Verify(verifier)) {
+    return false;
+  }
+
+  if (!response->objects()) {
+    return false;
+  }
+
+  for (const Object* object : *response->objects()) {
+    if (!object || !object->Verify(verifier)) {
+      return false;
+    }
+
+    const ObjectId* id = object->id();
+    if (!id || !id->Verify(verifier) || !id->digest()) {
+      return false;
+    }
+
+    const Data* data = object->data();
+    // No data is a valid response: it means the object was not found.
+    if (data && (!data->Verify(verifier) || !data->bytes())) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+bool IsValidResponse(flatbuffers::Verifier& verifier,
+                     const Response* response) {
+  if (!response) {
+    return false;
+  }
+
+  if (!response->Verify(verifier)) {
+    return false;
+  }
+
+  const NamespacePageId* namespace_page_id = response->namespace_page();
+  if (!namespace_page_id || !namespace_page_id->Verify(verifier)) {
+    return false;
+  }
+
+  if (!namespace_page_id->namespace_id() || !namespace_page_id->page_id()) {
+    return false;
+  }
+
+  switch (response->response_type()) {
+    case ResponseMessage_NONE:
+      // Response returned in case of unknown namespace or page.
+      return true;
+    case ResponseMessage_CommitResponse:
+      return IsValidCommitResponse(
+          verifier, static_cast<const CommitResponse*>(response->response()));
+    case ResponseMessage_ObjectResponse:
+      return IsValidObjectResponse(
+          verifier, static_cast<const ObjectResponse*>(response->response()));
+  }
+}
+
+}  // namespace
+
+const Message* ParseMessage(convert::ExtendedStringView data) {
+  flatbuffers::Verifier verifier(
+      reinterpret_cast<const unsigned char*>(data.data()), data.size());
+
+  if (!VerifyMessageBuffer(verifier)) {
+    return nullptr;
+  }
+
+  const Message* message = GetMessage(data.data());
+
+  if (!message || !message->Verify(verifier)) {
+    return nullptr;
+  }
+
+  switch (message->message_type()) {
+    case MessageUnion_NONE:
+      return nullptr;
+    case MessageUnion_Request:
+      if (!IsValidRequest(verifier,
+                          static_cast<const Request*>(message->message()))) {
+        return nullptr;
+      }
+      break;
+    case MessageUnion_Response:
+      if (!IsValidResponse(verifier,
+                           static_cast<const Response*>(message->message()))) {
+        return nullptr;
+      }
+      break;
+  }
+
+  return message;
+}
+
+}  // namespace p2p_sync
diff --git a/src/ledger/bin/p2p_sync/impl/encoding.h b/src/ledger/bin/p2p_sync/impl/encoding.h
new file mode 100644
index 0000000..aedf805
--- /dev/null
+++ b/src/ledger/bin/p2p_sync/impl/encoding.h
@@ -0,0 +1,19 @@
+// Copyright 2019 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef SRC_LEDGER_BIN_P2P_SYNC_IMPL_ENCODING_H_
+#define SRC_LEDGER_BIN_P2P_SYNC_IMPL_ENCODING_H_
+
+#include "peridot/lib/convert/convert.h"
+#include "src/ledger/bin/p2p_sync/impl/message_generated.h"
+
+namespace p2p_sync {
+
+// Parses |data| into a |Message| object. Returns nullptr if the message is
+// malformed. The returned pointer is valid as long as |data|'s data is valid.
+const Message* ParseMessage(convert::ExtendedStringView data);
+
+}  // namespace p2p_sync
+
+#endif  // SRC_LEDGER_BIN_P2P_SYNC_IMPL_ENCODING_H_
diff --git a/src/ledger/bin/p2p_sync/impl/encoding_unittest.cc b/src/ledger/bin/p2p_sync/impl/encoding_unittest.cc
new file mode 100644
index 0000000..a76bee9
--- /dev/null
+++ b/src/ledger/bin/p2p_sync/impl/encoding_unittest.cc
@@ -0,0 +1,83 @@
+// Copyright 2019 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "src/ledger/bin/p2p_sync/impl/encoding.h"
+
+#include "gtest/gtest.h"
+#include "message_generated.h"
+#include "peridot/lib/convert/convert.h"
+#include "src/ledger/bin/p2p_sync/impl/flatbuffer_message_factory.h"
+#include "src/lib/fxl/macros.h"
+
+namespace p2p_sync {
+namespace {
+
+TEST(ParseMessageTest, Valid_WatchStart) {
+  flatbuffers::FlatBufferBuilder buffer;
+  flatbuffers::Offset<WatchStartRequest> watch_start =
+      CreateWatchStartRequest(buffer);
+  flatbuffers::Offset<NamespacePageId> namespace_page_id =
+      CreateNamespacePageId(
+          buffer, convert::ToFlatBufferVector(&buffer, "namespace_id"),
+          convert::ToFlatBufferVector(&buffer, "page_id"));
+  flatbuffers::Offset<Request> request =
+      CreateRequest(buffer, namespace_page_id, RequestMessage_WatchStartRequest,
+                    watch_start.Union());
+  flatbuffers::Offset<Message> message =
+      CreateMessage(buffer, MessageUnion_Request, request.Union());
+  buffer.Finish(message);
+  const Message* message_ptr = ParseMessage(convert::ToStringView(buffer));
+  EXPECT_NE(message_ptr, nullptr);
+}
+
+TEST(ParseMessageTest, Invalid_NoNamespace) {
+  flatbuffers::FlatBufferBuilder buffer;
+  flatbuffers::Offset<WatchStartRequest> watch_start =
+      CreateWatchStartRequest(buffer);
+  flatbuffers::Offset<NamespacePageId> namespace_page_id;
+  flatbuffers::Offset<Request> request =
+      CreateRequest(buffer, namespace_page_id, RequestMessage_WatchStartRequest,
+                    watch_start.Union());
+  flatbuffers::Offset<Message> message =
+      CreateMessage(buffer, MessageUnion_Request, request.Union());
+  buffer.Finish(message);
+  const Message* message_ptr = ParseMessage(convert::ToStringView(buffer));
+  EXPECT_EQ(message_ptr, nullptr);
+}
+
+TEST(ParseMessageTest, Valid_ObjectResponseNoObject) {
+  flatbuffers::FlatBufferBuilder buffer;
+  flatbuffers::Offset<NamespacePageId> namespace_page_id =
+      CreateNamespacePageId(
+          buffer, convert::ToFlatBufferVector(&buffer, "namespace_id"),
+          convert::ToFlatBufferVector(&buffer, "page_id"));
+  flatbuffers::Offset<ObjectId> fb_object_id = CreateObjectId(
+      buffer, 1, 3, convert::ToFlatBufferVector(&buffer, "digest"));
+  std::vector<flatbuffers::Offset<Object>> fb_objects;
+  fb_objects.emplace_back(
+      CreateObject(buffer, fb_object_id, ObjectStatus_UNKNOWN_OBJECT));
+  flatbuffers::Offset<ObjectResponse> object_response =
+      CreateObjectResponse(buffer, buffer.CreateVector(fb_objects));
+  flatbuffers::Offset<Response> response =
+      CreateResponse(buffer, ResponseStatus_OK, namespace_page_id,
+                     ResponseMessage_ObjectResponse, object_response.Union());
+  flatbuffers::Offset<Message> message =
+      CreateMessage(buffer, MessageUnion_Response, response.Union());
+  buffer.Finish(message);
+
+  const Message* message_ptr = ParseMessage(convert::ToStringView(buffer));
+  EXPECT_NE(message_ptr, nullptr);
+}
+
+TEST(ParseMessageTest, Valid_ResponseUnknownNamespace) {
+  flatbuffers::FlatBufferBuilder buffer;
+  CreateUnknownResponseMessage(&buffer, "namespace", "page",
+                               ResponseStatus_UNKNOWN_NAMESPACE);
+
+  const Message* message_ptr = ParseMessage(convert::ToStringView(buffer));
+  EXPECT_NE(message_ptr, nullptr);
+}
+
+}  // namespace
+}  // namespace p2p_sync
diff --git a/src/ledger/bin/p2p_sync/impl/flatbuffer_message_factory.h b/src/ledger/bin/p2p_sync/impl/flatbuffer_message_factory.h
index 6d85bce..c52b72f 100644
--- a/src/ledger/bin/p2p_sync/impl/flatbuffer_message_factory.h
+++ b/src/ledger/bin/p2p_sync/impl/flatbuffer_message_factory.h
@@ -12,6 +12,8 @@
 
 namespace p2p_sync {
 
+// Fill |buffer| with a new |Message| containing a |Response| for an unknown
+// namespace or page.
 void CreateUnknownResponseMessage(flatbuffers::FlatBufferBuilder* buffer,
                                   fxl::StringView namespace_id,
                                   fxl::StringView page_id,
diff --git a/src/ledger/bin/p2p_sync/impl/message_holder.h b/src/ledger/bin/p2p_sync/impl/message_holder.h
index add1a49..5f4bc43 100644
--- a/src/ledger/bin/p2p_sync/impl/message_holder.h
+++ b/src/ledger/bin/p2p_sync/impl/message_holder.h
@@ -8,9 +8,12 @@
 #include <lib/fit/function.h>
 
 #include <functional>
+#include <optional>
 #include <utility>
 #include <vector>
 
+#include "peridot/lib/convert/convert.h"
+#include "src/ledger/bin/p2p_sync/impl/encoding.h"
 #include "src/lib/fxl/strings/string_view.h"
 
 namespace p2p_sync {
@@ -20,14 +23,8 @@
 class MessageHolder {
  public:
   // Create a new MessageHolder from the provided data and a parser.
-  MessageHolder(fxl::StringView data,
-                fit::function<const M*(const uint8_t*)> get_message)
-      : data_(data.begin(), data.end()), message_(get_message(data_.data())) {}
-
-  // Create a new MessageHolder from the provided data and a parser.
-  MessageHolder(std::vector<uint8_t> data,
-                fit::function<const M*(const uint8_t*)> get_message)
-      : data_(std::move(data)), message_(get_message(data_.data())) {}
+  MessageHolder(std::unique_ptr<std::vector<uint8_t>> data, const M* message)
+      : data_(std::move(data)), message_(message) {}
 
   // Create a new MessageHolder from the current object and a function to
   // specialize the message. The current message holder is destroyed in the
@@ -71,10 +68,27 @@
     other.message_ = nullptr;
   }
 
-  std::vector<uint8_t> data_;
+  std::unique_ptr<std::vector<uint8_t>> data_;
   const M* message_;
 };
 
+// Creates a new MessageHolder to contained a message, or nullopt if no message
+// can be obtained.
+template <class M>
+std::optional<MessageHolder<M>> CreateMessageHolder(
+    fxl::StringView data,
+    fit::function<const M*(convert::ExtendedStringView)> get_message) {
+  std::unique_ptr<std::vector<uint8_t>> data_vec =
+      std::make_unique<std::vector<uint8_t>>(data.begin(), data.end());
+
+  const M* message = get_message(*data_vec);
+
+  if (!message) {
+    return std::nullopt;
+  }
+
+  return MessageHolder<M>(std::move(data_vec), message);
+}
 }  // namespace p2p_sync
 
 #endif  // SRC_LEDGER_BIN_P2P_SYNC_IMPL_MESSAGE_HOLDER_H_
diff --git a/src/ledger/bin/p2p_sync/impl/page_communicator_impl.cc b/src/ledger/bin/p2p_sync/impl/page_communicator_impl.cc
index 1f794d9..cf927cf 100644
--- a/src/ledger/bin/p2p_sync/impl/page_communicator_impl.cc
+++ b/src/ledger/bin/p2p_sync/impl/page_communicator_impl.cc
@@ -11,6 +11,7 @@
 #include "peridot/lib/convert/convert.h"
 #include "src/ledger/bin/p2p_sync/impl/message_generated.h"
 #include "src/ledger/bin/storage/public/read_data_source.h"
+#include "src/ledger/bin/storage/public/types.h"
 #include "src/ledger/lib/coroutine/coroutine_waiter.h"
 #include "src/lib/fxl/memory/ref_ptr.h"
 
@@ -391,12 +392,15 @@
 
 void PageCommunicatorImpl::BuildWatchStartBuffer(
     flatbuffers::FlatBufferBuilder* buffer) {
+  flatbuffers::Offset<WatchStartRequest> watch_start =
+      CreateWatchStartRequest(*buffer);
   flatbuffers::Offset<NamespacePageId> namespace_page_id =
       CreateNamespacePageId(*buffer,
                             convert::ToFlatBufferVector(buffer, namespace_id_),
                             convert::ToFlatBufferVector(buffer, page_id_));
-  flatbuffers::Offset<Request> request = CreateRequest(
-      *buffer, namespace_page_id, RequestMessage_WatchStartRequest);
+  flatbuffers::Offset<Request> request =
+      CreateRequest(*buffer, namespace_page_id,
+                    RequestMessage_WatchStartRequest, watch_start.Union());
   flatbuffers::Offset<Message> message =
       CreateMessage(*buffer, MessageUnion_Request, request.Union());
   buffer->Finish(message);
@@ -404,12 +408,15 @@
 
 void PageCommunicatorImpl::BuildWatchStopBuffer(
     flatbuffers::FlatBufferBuilder* buffer) {
+  flatbuffers::Offset<WatchStopRequest> watch_stop =
+      CreateWatchStopRequest(*buffer);
   flatbuffers::Offset<NamespacePageId> namespace_page_id =
       CreateNamespacePageId(*buffer,
                             convert::ToFlatBufferVector(buffer, namespace_id_),
                             convert::ToFlatBufferVector(buffer, page_id_));
-  flatbuffers::Offset<Request> request = CreateRequest(
-      *buffer, namespace_page_id, RequestMessage_WatchStopRequest);
+  flatbuffers::Offset<Request> request =
+      CreateRequest(*buffer, namespace_page_id, RequestMessage_WatchStopRequest,
+                    watch_stop.Union());
   flatbuffers::Offset<Message> message =
       CreateMessage(*buffer, MessageUnion_Request, request.Union());
   buffer->Finish(message);
diff --git a/src/ledger/bin/p2p_sync/impl/page_communicator_impl_unittest.cc b/src/ledger/bin/p2p_sync/impl/page_communicator_impl_unittest.cc
index 2c6c52f..ca4e81a 100644
--- a/src/ledger/bin/p2p_sync/impl/page_communicator_impl_unittest.cc
+++ b/src/ledger/bin/p2p_sync/impl/page_communicator_impl_unittest.cc
@@ -22,6 +22,8 @@
 #include "gmock/gmock.h"
 #include "peridot/lib/convert/convert.h"
 #include "src/ledger/bin/p2p_sync/impl/device_mesh.h"
+#include "src/ledger/bin/p2p_sync/impl/encoding.h"
+#include "src/ledger/bin/p2p_sync/impl/message_generated.h"
 #include "src/ledger/bin/storage/fake/fake_object.h"
 #include "src/ledger/bin/storage/testing/commit_empty_impl.h"
 #include "src/ledger/bin/storage/testing/page_storage_empty_impl.h"
@@ -202,12 +204,15 @@
 void BuildWatchStartBuffer(flatbuffers::FlatBufferBuilder* buffer,
                            fxl::StringView namespace_id,
                            fxl::StringView page_id) {
+  flatbuffers::Offset<WatchStartRequest> watch_start =
+      CreateWatchStartRequest(*buffer);
   flatbuffers::Offset<NamespacePageId> namespace_page_id =
       CreateNamespacePageId(*buffer,
                             convert::ToFlatBufferVector(buffer, namespace_id),
                             convert::ToFlatBufferVector(buffer, page_id));
-  flatbuffers::Offset<Request> request = CreateRequest(
-      *buffer, namespace_page_id, RequestMessage_WatchStartRequest);
+  flatbuffers::Offset<Request> request =
+      CreateRequest(*buffer, namespace_page_id,
+                    RequestMessage_WatchStartRequest, watch_start.Union());
   flatbuffers::Offset<Message> message =
       CreateMessage(*buffer, MessageUnion_Request, request.Union());
   buffer->Finish(message);
@@ -216,12 +221,15 @@
 void BuildWatchStopBuffer(flatbuffers::FlatBufferBuilder* buffer,
                           fxl::StringView namespace_id,
                           fxl::StringView page_id) {
+  flatbuffers::Offset<WatchStopRequest> watch_stop =
+      CreateWatchStopRequest(*buffer);
   flatbuffers::Offset<NamespacePageId> namespace_page_id =
       CreateNamespacePageId(*buffer,
                             convert::ToFlatBufferVector(buffer, namespace_id),
                             convert::ToFlatBufferVector(buffer, page_id));
-  flatbuffers::Offset<Request> request = CreateRequest(
-      *buffer, namespace_page_id, RequestMessage_WatchStopRequest);
+  flatbuffers::Offset<Request> request =
+      CreateRequest(*buffer, namespace_page_id, RequestMessage_WatchStopRequest,
+                    watch_stop.Union());
   flatbuffers::Offset<Message> message =
       CreateMessage(*buffer, MessageUnion_Request, request.Union());
   buffer->Finish(message);
@@ -409,8 +417,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> new_device_message(convert::ToStringView(buffer),
-                                            &GetMessage);
+  MessageHolder<Message> new_device_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(new_device_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -464,8 +472,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> new_device_message(convert::ToStringView(buffer),
-                                            &GetMessage);
+  MessageHolder<Message> new_device_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   // If storage fails to mark the page as synced to a peer, the mesh should not
   // be updated.
   storage.mark_synced_to_peer_status = storage::Status::IO_ERROR;
@@ -502,8 +510,8 @@
   BuildObjectRequestBuffer(&request_buffer, "ledger", "page",
                            {MakeObjectIdentifier("object_digest"),
                             MakeObjectIdentifier("object_digest2")});
-  MessageHolder<Message> request_message(convert::ToStringView(request_buffer),
-                                         &GetMessage);
+  MessageHolder<Message> request_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(request_buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(request_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -558,8 +566,8 @@
   flatbuffers::FlatBufferBuilder request_buffer;
   BuildObjectRequestBuffer(&request_buffer, "ledger", "page",
                            {MakeObjectIdentifier("object_digest")});
-  MessageHolder<Message> request_message(convert::ToStringView(request_buffer),
-                                         &GetMessage);
+  MessageHolder<Message> request_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(request_buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(request_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -607,8 +615,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> new_device_message(convert::ToStringView(buffer),
-                                            &GetMessage);
+  MessageHolder<Message> new_device_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(new_device_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -635,8 +643,8 @@
       &response_buffer, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "foo_data", false),
        std::make_tuple(MakeObjectIdentifier("bar"), "bar_data", false)});
-  MessageHolder<Message> response_message(
-      convert::ToStringView(response_buffer), &GetMessage);
+  MessageHolder<Message> response_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer), &ParseMessage);
   page_communicator.OnNewResponse(
       "device2", std::move(response_message)
                      .TakeAndMap<Response>([](const Message* message) {
@@ -658,8 +666,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> new_device_message(convert::ToStringView(buffer),
-                                            &GetMessage);
+  MessageHolder<Message> new_device_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(new_device_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -685,8 +693,8 @@
   BuildObjectResponseBuffer(
       &response_buffer, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "foo_data", true)});
-  MessageHolder<Message> response_message(
-      convert::ToStringView(response_buffer), &GetMessage);
+  MessageHolder<Message> response_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer), &ParseMessage);
   page_communicator.OnNewResponse(
       "device2", std::move(response_message)
                      .TakeAndMap<Response>([](const Message* message) {
@@ -708,7 +716,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> message(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2",
       std::move(message).TakeAndMap<Request>([](const Message* message) {
@@ -734,8 +743,8 @@
   BuildObjectResponseBuffer(
       &response_buffer, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "", false)});
-  MessageHolder<Message> response_message(
-      convert::ToStringView(response_buffer), &GetMessage);
+  MessageHolder<Message> response_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer), &ParseMessage);
   page_communicator.OnNewResponse(
       "device2", std::move(response_message)
                      .TakeAndMap<Response>([](const Message* message) {
@@ -756,16 +765,18 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> message(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2",
       std::move(message).TakeAndMap<Request>([](const Message* message) {
         return static_cast<const Request*>(message->message());
       }));
-  message = MessageHolder<Message>(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message_copy = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device3",
-      std::move(message).TakeAndMap<Request>([](const Message* message) {
+      std::move(message_copy).TakeAndMap<Request>([](const Message* message) {
         return static_cast<const Request*>(message->message());
       }));
 
@@ -786,8 +797,8 @@
   BuildObjectResponseBuffer(
       &response_buffer_1, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "", false)});
-  MessageHolder<Message> message_1(convert::ToStringView(response_buffer_1),
-                                   &GetMessage);
+  MessageHolder<Message> message_1 = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer_1), &ParseMessage);
   page_communicator.OnNewResponse(
       "device2",
       std::move(message_1).TakeAndMap<Response>([](const Message* message) {
@@ -799,8 +810,8 @@
   BuildObjectResponseBuffer(
       &response_buffer_2, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "foo_data", false)});
-  MessageHolder<Message> message_2(convert::ToStringView(response_buffer_2),
-                                   &GetMessage);
+  MessageHolder<Message> message_2 = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer_2), &ParseMessage);
   page_communicator.OnNewResponse(
       "device3",
       std::move(message_2).TakeAndMap<Response>([](const Message* message) {
@@ -823,16 +834,18 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> message(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2",
       std::move(message).TakeAndMap<Request>([](const Message* message) {
         return static_cast<const Request*>(message->message());
       }));
-  message = MessageHolder<Message>(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message_copy = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device3",
-      std::move(message).TakeAndMap<Request>([](const Message* message) {
+      std::move(message_copy).TakeAndMap<Request>([](const Message* message) {
         return static_cast<const Request*>(message->message());
       }));
 
@@ -853,8 +866,8 @@
   BuildObjectResponseBuffer(
       &response_buffer_1, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "", false)});
-  MessageHolder<Message> message_1(convert::ToStringView(response_buffer_1),
-                                   &GetMessage);
+  MessageHolder<Message> message_1 = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer_1), &ParseMessage);
   page_communicator.OnNewResponse(
       "device2",
       std::move(message_1).TakeAndMap<Response>([](const Message* message) {
@@ -866,8 +879,8 @@
   BuildObjectResponseBuffer(
       &response_buffer_2, "ledger", "page",
       {std::make_tuple(MakeObjectIdentifier("foo"), "", false)});
-  MessageHolder<Message> message_2(convert::ToStringView(response_buffer_2),
-                                   &GetMessage);
+  MessageHolder<Message> message_2 = *CreateMessageHolder<Message>(
+      convert::ToStringView(response_buffer_2), &ParseMessage);
   page_communicator.OnNewResponse(
       "device3",
       std::move(message_2).TakeAndMap<Response>([](const Message* message) {
@@ -888,7 +901,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> message(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator_1.OnNewRequest(
       "device2",
       std::move(message).TakeAndMap<Request>([](const Message* message) {
@@ -924,12 +938,8 @@
   ASSERT_EQ(1u, mesh.messages_.size());
   EXPECT_EQ("device2", mesh.messages_[0].first);
 
-  flatbuffers::Verifier verifier(
-      reinterpret_cast<const unsigned char*>(mesh.messages_[0].second.data()),
-      mesh.messages_[0].second.size());
-  ASSERT_TRUE(VerifyMessageBuffer(verifier));
-
-  MessageHolder<Message> reply_message(mesh.messages_[0].second, &GetMessage);
+  MessageHolder<Message> reply_message =
+      *CreateMessageHolder<Message>(mesh.messages_[0].second, &ParseMessage);
   ASSERT_EQ(MessageUnion_Response, reply_message->message_type());
   MessageHolder<Response> response =
       std::move(reply_message).TakeAndMap<Response>([](const Message* message) {
@@ -969,7 +979,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> message(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2",
       std::move(message).TakeAndMap<Request>([](const Message* message) {
@@ -1007,8 +1018,8 @@
 
   flatbuffers::FlatBufferBuilder stop_buffer;
   BuildWatchStopBuffer(&stop_buffer, "ledger", "page");
-  MessageHolder<Message> watch_stop_message(convert::ToStringView(stop_buffer),
-                                            &GetMessage);
+  MessageHolder<Message> watch_stop_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(stop_buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(watch_stop_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -1053,8 +1064,8 @@
   BuildCommitRequestBuffer(&request_buffer, "ledger", "page",
                            {storage::CommitId(commit_1.GetId()),
                             storage::CommitId("missing_commit")});
-  MessageHolder<Message> request_message(convert::ToStringView(request_buffer),
-                                         &GetMessage);
+  MessageHolder<Message> request_message = *CreateMessageHolder<Message>(
+      convert::ToStringView(request_buffer), &ParseMessage);
   page_communicator.OnNewRequest(
       "device2", std::move(request_message)
                      .TakeAndMap<Request>([](const Message* message) {
@@ -1106,7 +1117,8 @@
 
   flatbuffers::FlatBufferBuilder buffer;
   BuildWatchStartBuffer(&buffer, "ledger", "page");
-  MessageHolder<Message> message(convert::ToStringView(buffer), &GetMessage);
+  MessageHolder<Message> message = *CreateMessageHolder<Message>(
+      convert::ToStringView(buffer), &ParseMessage);
   page_communicator_1.OnNewRequest(
       "device2",
       std::move(message).TakeAndMap<Request>([](const Message* message) {
@@ -1133,12 +1145,8 @@
   EXPECT_EQ("device2", mesh.messages_[0].first);
 
   {
-    flatbuffers::Verifier verifier(
-        reinterpret_cast<const unsigned char*>(mesh.messages_[0].second.data()),
-        mesh.messages_[0].second.size());
-    ASSERT_TRUE(VerifyMessageBuffer(verifier));
-
-    MessageHolder<Message> reply_message(mesh.messages_[0].second, &GetMessage);
+    MessageHolder<Message> reply_message =
+        *CreateMessageHolder<Message>(mesh.messages_[0].second, &ParseMessage);
     ASSERT_EQ(MessageUnion_Response, reply_message->message_type());
     MessageHolder<Response> response =
         std::move(reply_message)
@@ -1170,13 +1178,8 @@
   EXPECT_EQ("device1", mesh.messages_[1].first);
 
   {
-    flatbuffers::Verifier verifier(
-        reinterpret_cast<const unsigned char*>(mesh.messages_[1].second.data()),
-        mesh.messages_[0].second.size());
-    ASSERT_TRUE(VerifyMessageBuffer(verifier));
-
-    MessageHolder<Message> request_message(mesh.messages_[1].second,
-                                           &GetMessage);
+    MessageHolder<Message> request_message =
+        *CreateMessageHolder<Message>(mesh.messages_[1].second, &ParseMessage);
     ASSERT_EQ(MessageUnion_Request, request_message->message_type());
     MessageHolder<Request> request =
         std::move(request_message)
@@ -1201,12 +1204,8 @@
   EXPECT_EQ("device2", mesh.messages_[2].first);
 
   {
-    flatbuffers::Verifier verifier(
-        reinterpret_cast<const unsigned char*>(mesh.messages_[2].second.data()),
-        mesh.messages_[0].second.size());
-    ASSERT_TRUE(VerifyMessageBuffer(verifier));
-
-    MessageHolder<Message> reply_message(mesh.messages_[2].second, &GetMessage);
+    MessageHolder<Message> reply_message =
+        *CreateMessageHolder<Message>(mesh.messages_[2].second, &ParseMessage);
     ASSERT_EQ(MessageUnion_Response, reply_message->message_type());
     MessageHolder<Response> response =
         std::move(reply_message)
diff --git a/src/ledger/bin/p2p_sync/impl/user_communicator_impl.cc b/src/ledger/bin/p2p_sync/impl/user_communicator_impl.cc
index 78b80c4..1e20343a 100644
--- a/src/ledger/bin/p2p_sync/impl/user_communicator_impl.cc
+++ b/src/ledger/bin/p2p_sync/impl/user_communicator_impl.cc
@@ -5,6 +5,7 @@
 #include "src/ledger/bin/p2p_sync/impl/user_communicator_impl.h"
 
 #include "peridot/lib/ledger_client/constants.h"
+#include "src/ledger/bin/p2p_sync/impl/encoding.h"
 #include "src/ledger/bin/p2p_sync/impl/flatbuffer_message_factory.h"
 #include "src/ledger/bin/p2p_sync/impl/ledger_communicator_impl.h"
 #include "src/ledger/bin/p2p_sync/impl/message_generated.h"
@@ -48,16 +49,15 @@
 
 void UserCommunicatorImpl::OnNewMessage(fxl::StringView source,
                                         fxl::StringView data) {
-  flatbuffers::Verifier verifier(
-      reinterpret_cast<const unsigned char*>(data.data()), data.size());
-  if (!VerifyMessageBuffer(verifier)) {
+  std::optional<MessageHolder<Message>> message =
+      CreateMessageHolder<Message>(data, &ParseMessage);
+  if (!message) {
     // Wrong serialization, abort.
     FXL_LOG(ERROR) << "The message received is malformed.";
     return;
   };
-  MessageHolder<Message> message(data, &GetMessage);
-  const NamespacePageId* namespace_page_id;
-  switch (message->message_type()) {
+
+  switch ((*message)->message_type()) {
     case MessageUnion_NONE:
       FXL_LOG(ERROR) << "The message received is unexpected at this point.";
       return;
@@ -65,10 +65,10 @@
 
     case MessageUnion_Request: {
       MessageHolder<Request> request =
-          std::move(message).TakeAndMap<Request>([](const Message* message) {
+          std::move(*message).TakeAndMap<Request>([](const Message* message) {
             return static_cast<const Request*>(message->message());
           });
-      namespace_page_id = request->namespace_page();
+      const NamespacePageId* namespace_page_id = request->namespace_page();
 
       std::string namespace_id(namespace_page_id->namespace_id()->begin(),
                                namespace_page_id->namespace_id()->end());
@@ -89,10 +89,10 @@
 
     case MessageUnion_Response: {
       MessageHolder<Response> response =
-          std::move(message).TakeAndMap<Response>([](const Message* message) {
+          std::move(*message).TakeAndMap<Response>([](const Message* message) {
             return static_cast<const Response*>(message->message());
           });
-      namespace_page_id = response->namespace_page();
+      const NamespacePageId* namespace_page_id = response->namespace_page();
       std::string namespace_id(namespace_page_id->namespace_id()->begin(),
                                namespace_page_id->namespace_id()->end());
       std::string page_id(namespace_page_id->page_id()->begin(),
@@ -100,10 +100,10 @@
 
       const auto& it = ledgers_.find(namespace_id);
       if (it == ledgers_.end()) {
-        // We are receiving a response for a ledger that no longer exists. This
-        // can happen in normal operation, and we cannot do anything with this
-        // message: we can't send it to a ledger, and we don't send responses to
-        // responses. So we just drop it here.
+        // We are receiving a response for a ledger that no longer exists.
+        // This can happen in normal operation, and we cannot do anything with
+        // this message: we can't send it to a ledger, and we don't send
+        // responses to responses. So we just drop it here.
         return;
       }
       it->second->OnNewResponse(source, page_id, std::move(response));
diff --git a/src/ledger/bin/p2p_sync/impl/user_communicator_impl_fuzztest.cc b/src/ledger/bin/p2p_sync/impl/user_communicator_impl_fuzztest.cc
new file mode 100644
index 0000000..9368b69
--- /dev/null
+++ b/src/ledger/bin/p2p_sync/impl/user_communicator_impl_fuzztest.cc
@@ -0,0 +1,74 @@
+// Copyright 2019 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <lib/fidl/cpp/binding.h>
+#include <lib/fit/function.h>
+
+#include <algorithm>
+#include <string>
+
+#include "src/ledger/bin/p2p_provider/impl/p2p_provider_impl.h"
+#include "src/ledger/bin/p2p_provider/public/p2p_provider.h"
+#include "src/ledger/bin/p2p_sync/impl/user_communicator_impl.h"
+#include "src/ledger/bin/storage/public/page_sync_client.h"
+#include "src/ledger/bin/storage/testing/page_storage_empty_impl.h"
+#include "src/ledger/lib/coroutine/coroutine_impl.h"
+#include "src/lib/fxl/macros.h"
+
+namespace p2p_sync {
+namespace {
+
+class TestPageStorage : public storage::PageStorageEmptyImpl {
+ public:
+  TestPageStorage() = default;
+  ~TestPageStorage() {}
+
+  storage::PageId GetId() override { return "page"; }
+
+  void SetSyncDelegate(storage::PageSyncDelegate* page_sync) override {
+    return;
+  }
+};
+
+class FuzzingP2PProvider : public p2p_provider::P2PProvider {
+ public:
+  FuzzingP2PProvider() = default;
+
+  void Start(Client* client) override { client_ = client; }
+
+  bool SendMessage(fxl::StringView destination, fxl::StringView data) override {
+    FXL_NOTIMPLEMENTED();
+    return false;
+  }
+
+  Client* client_;
+};
+
+// Fuzz the peer-to-peer messages received by a |UserCommunicatorImpl|.
+extern "C" int LLVMFuzzerTestOneInput(const uint8_t* Data, size_t Size) {
+  std::string bytes(reinterpret_cast<const char*>(Data), Size);
+
+  coroutine::CoroutineServiceImpl coroutine_service;
+  auto provider = std::make_unique<FuzzingP2PProvider>();
+  FuzzingP2PProvider* provider_ptr = provider.get();
+
+  UserCommunicatorImpl user_communicator(std::move(provider),
+                                         &coroutine_service);
+  user_communicator.Start();
+  auto ledger_communicator = user_communicator.GetLedgerCommunicator("ledger");
+
+  storage::PageStorageEmptyImpl page_storage;
+
+  auto page_communicator =
+      ledger_communicator->GetPageCommunicator(&page_storage, &page_storage);
+
+  provider_ptr->client_->OnDeviceChange("device",
+                                        p2p_provider::DeviceChangeType::NEW);
+  provider_ptr->client_->OnNewMessage("device", bytes);
+
+  return 0;
+}
+
+}  // namespace
+}  // namespace p2p_sync