CBG-3332 increase log information for documents not imported

db/import_listener.go

@@ -150,6 +150,7 @@ func (il *importListener) ProcessFeedEvent(event sgbucket.FeedEvent) (shouldPers
 
 	// If this is a binary document we can ignore, but update checkpoint to avoid reprocessing upon restart
 	if event.DataType == base.MemcachedDataTypeRaw {
+		base.InfofCtx(il.loggingCtx, base.KeyImport, "Ignoring binary mutation event for %s.", base.UD(event.Key))
 		return true
 	}
 
@@ -179,10 +180,12 @@ func (il *importListener) ImportFeedEvent(event sgbucket.FeedEvent) {
 
 	syncData, rawBody, rawXattr, rawUserXattr, err := UnmarshalDocumentSyncDataFromFeed(event.Value, event.DataType, collectionCtx.userXattrKey(), false)
 	if err != nil {
-		base.DebugfCtx(il.loggingCtx, base.KeyImport, "Found sync metadata, but unable to unmarshal for feed document %q.  Will not be imported.  Error: %v", base.UD(event.Key), err)
 		if err == base.ErrEmptyMetadata {
 			base.WarnfCtx(il.loggingCtx, "Unexpected empty metadata when processing feed event.  docid: %s opcode: %v datatype:%v", base.UD(event.Key), event.Opcode, event.DataType)
+		} else {
+			base.WarnfCtx(il.loggingCtx, "Found sync metadata, but unable to unmarshal for feed document %q.  Will not be imported.  Error: %v", base.UD(event.Key), err)
 		}
+		il.importStats.ImportErrorCount.Add(1)
 		return
 	}
 

db/import_test.go

@@ -557,5 +557,29 @@ func TestImportNonZeroStart(t *testing.T) {
 	doc, err := collection.GetDocument(base.TestCtx(t), doc1, DocUnmarshalAll)
 	require.NoError(t, err)
 	require.Equal(t, revID1, doc.SyncData.CurrentRev)
+}
+
+// TestImportInvalidMetadata tests triggering an import error if the metadata is unmarshalable
+func TestImportInvalidMetadata(t *testing.T) {
+	base.SkipImportTestsIfNotEnabled(t)
+	bucket := base.GetTestBucket(t)
+	defer bucket.Close()
+
+	db, ctx := setupTestDBWithOptionsAndImport(t, bucket, DatabaseContextOptions{})
+	defer db.Close(ctx)
 
+	// make sure no documents are imported
+	require.Equal(t, int64(0), db.DbStats.SharedBucketImport().ImportCount.Value())
+	require.Equal(t, int64(0), db.DbStats.SharedBucketImport().ImportErrorCount.Value())
+
+	// write a document with inline sync metadata that is unmarshalable, triggering an import error
+	// can't write a document with invalid sync metadata as an xattr, so rely on legacy behavior
+	_, err := bucket.GetSingleDataStore().Add("doc1", 0, `{"foo" : "bar", "_sync" : 1 }`)
+	require.NoError(t, err)
+
+	_, ok := base.WaitForStat(func() int64 {
+		return db.DbStats.SharedBucketImport().ImportErrorCount.Value()
+	}, 1)
+	require.True(t, ok)
+	require.Equal(t, int64(0), db.DbStats.SharedBucketImport().ImportCount.Value())
 }