diff --git a/.changeset/seven-crabs-join.md b/.changeset/seven-crabs-join.md
new file mode 100644
index 00000000000..a845151cc84
--- /dev/null
+++ b/.changeset/seven-crabs-join.md
@@ -0,0 +1,2 @@
+---
+---
diff --git a/packages/firestore/exp/index.d.ts b/packages/firestore/exp/index.d.ts
index 1464cb1c842..86364a431f6 100644
--- a/packages/firestore/exp/index.d.ts
+++ b/packages/firestore/exp/index.d.ts
@@ -362,6 +362,8 @@ export function updateDoc(
 ): Promise<void>;
 export function deleteDoc(reference: DocumentReference<unknown>): Promise<void>;
 
+// TODO(firestoreexp): Update API Proposal to use FirestoreError in these
+// callbacks
 export function onSnapshot<T>(
   reference: DocumentReference<T>,
   observer: {
@@ -375,28 +377,28 @@ export function onSnapshot<T>(
   options: SnapshotListenOptions,
   observer: {
     next?: (snapshot: DocumentSnapshot<T>) => void;
-    error?: (error: Error) => void;
+    error?: (error: FirestoreError) => void;
     complete?: () => void;
   }
 ): () => void;
 export function onSnapshot<T>(
   reference: DocumentReference<T>,
   onNext: (snapshot: DocumentSnapshot<T>) => void,
-  onError?: (error: Error) => void,
+  onError?: (error: FirestoreError) => void,
   onCompletion?: () => void
 ): () => void;
 export function onSnapshot<T>(
   reference: DocumentReference<T>,
   options: SnapshotListenOptions,
   onNext: (snapshot: DocumentSnapshot<T>) => void,
-  onError?: (error: Error) => void,
+  onError?: (error: FirestoreError) => void,
   onCompletion?: () => void
 ): () => void;
 export function onSnapshot<T>(
   query: Query<T>,
   observer: {
     next?: (snapshot: QuerySnapshot<T>) => void;
-    error?: (error: Error) => void;
+    error?: (error: FirestoreError) => void;
     complete?: () => void;
   }
 ): () => void;
@@ -405,28 +407,28 @@ export function onSnapshot<T>(
   options: SnapshotListenOptions,
   observer: {
     next?: (snapshot: QuerySnapshot<T>) => void;
-    error?: (error: Error) => void;
+    error?: (error: FirestoreError) => void;
     complete?: () => void;
   }
 ): () => void;
 export function onSnapshot<T>(
   query: Query<T>,
   onNext: (snapshot: QuerySnapshot<T>) => void,
-  onError?: (error: Error) => void,
+  onError?: (error: FirestoreError) => void,
   onCompletion?: () => void
 ): () => void;
 export function onSnapshot<T>(
   query: Query<T>,
   options: SnapshotListenOptions,
   onNext: (snapshot: QuerySnapshot<T>) => void,
-  onError?: (error: Error) => void,
+  onError?: (error: FirestoreError) => void,
   onCompletion?: () => void
 ): () => void;
 export function onSnapshotsInSync(
   firestore: FirebaseFirestore,
   observer: {
     next?: (value: void) => void;
-    error?: (error: Error) => void;
+    error?: (error: FirestoreError) => void;
     complete?: () => void;
   }
 ): () => void;
diff --git a/packages/firestore/exp/index.node.ts b/packages/firestore/exp/index.node.ts
index ac70fdf22b7..5113219901e 100644
--- a/packages/firestore/exp/index.node.ts
+++ b/packages/firestore/exp/index.node.ts
@@ -49,7 +49,16 @@ export {
 
 export { runTransaction, Transaction } from '../lite/src/api/transaction';
 
-export { getDoc, getDocFromCache, getDocFromServer } from './src/api/reference';
+export {
+  getDoc,
+  getDocFromCache,
+  getDocFromServer,
+  onSnapshot,
+  setDoc,
+  updateDoc,
+  deleteDoc,
+  addDoc
+} from './src/api/reference';
 
 export {
   FieldValue,
diff --git a/packages/firestore/exp/src/api/reference.ts b/packages/firestore/exp/src/api/reference.ts
index e769ff2174c..bbc30bd0449 100644
--- a/packages/firestore/exp/src/api/reference.ts
+++ b/packages/firestore/exp/src/api/reference.ts
@@ -28,10 +28,13 @@ import { debugAssert } from '../../../src/util/assert';
 import { cast } from '../../../lite/src/api/util';
 import { DocumentSnapshot, QuerySnapshot } from './snapshot';
 import {
+  addDocSnapshotListener,
+  addQuerySnapshotListener,
   applyFirestoreDataConverter,
   getDocsViaSnapshotListener,
   getDocViaSnapshotListener,
-  SnapshotMetadata
+  SnapshotMetadata,
+  validateHasExplicitOrderByForLimitToLast
 } from '../../../src/api/database';
 import { ViewSnapshot } from '../../../src/core/view_snapshot';
 import {
@@ -44,6 +47,14 @@ import {
 import { Document } from '../../../src/model/document';
 import { DeleteMutation, Precondition } from '../../../src/model/mutation';
 import { FieldPath } from '../../../src/api/field_path';
+import {
+  CompleteFn,
+  ErrorFn,
+  isPartialObserver,
+  NextFn,
+  PartialObserver,
+  Unsubscribe
+} from '../../../src/api/observer';
 
 export function getDoc<T>(
   reference: firestore.DocumentReference<T>
@@ -101,17 +112,14 @@ export function getQuery<T>(
 ): Promise<QuerySnapshot<T>> {
   const internalQuery = cast<Query<T>>(query, Query);
   const firestore = cast<Firestore>(query.firestore, Firestore);
+
+  validateHasExplicitOrderByForLimitToLast(internalQuery._query);
   return firestore._getFirestoreClient().then(async firestoreClient => {
     const snapshot = await getDocsViaSnapshotListener(
       firestoreClient,
       internalQuery._query
     );
-    return new QuerySnapshot(
-      firestore,
-      internalQuery,
-      snapshot,
-      new SnapshotMetadata(snapshot.hasPendingWrites, snapshot.fromCache)
-    );
+    return new QuerySnapshot(firestore, internalQuery, snapshot);
   });
 }
 
@@ -124,12 +132,7 @@ export function getQueryFromCache<T>(
     const snapshot = await firestoreClient.getDocumentsFromLocalCache(
       internalQuery._query
     );
-    return new QuerySnapshot(
-      firestore,
-      internalQuery,
-      snapshot,
-      new SnapshotMetadata(snapshot.hasPendingWrites, /* fromCache= */ true)
-    );
+    return new QuerySnapshot(firestore, internalQuery, snapshot);
   });
 }
 
@@ -144,12 +147,7 @@ export function getQueryFromServer<T>(
       internalQuery._query,
       { source: 'server' }
     );
-    return new QuerySnapshot(
-      firestore,
-      internalQuery,
-      snapshot,
-      new SnapshotMetadata(snapshot.hasPendingWrites, snapshot.fromCache)
-    );
+    return new QuerySnapshot(firestore, internalQuery, snapshot);
   });
 }
 
@@ -280,6 +278,159 @@ export function addDoc<T>(
     .then(() => docRef);
 }
 
+// TODO(firestorexp): Make sure these overloads are tested via the Firestore
+// integration tests
+export function onSnapshot<T>(
+  reference: firestore.DocumentReference<T>,
+  observer: {
+    next?: (snapshot: firestore.DocumentSnapshot<T>) => void;
+    error?: (error: firestore.FirestoreError) => void;
+    complete?: () => void;
+  }
+): Unsubscribe;
+export function onSnapshot<T>(
+  reference: firestore.DocumentReference<T>,
+  options: firestore.SnapshotListenOptions,
+  observer: {
+    next?: (snapshot: firestore.DocumentSnapshot<T>) => void;
+    error?: (error: firestore.FirestoreError) => void;
+    complete?: () => void;
+  }
+): Unsubscribe;
+export function onSnapshot<T>(
+  reference: firestore.DocumentReference<T>,
+  onNext: (snapshot: firestore.DocumentSnapshot<T>) => void,
+  onError?: (error: firestore.FirestoreError) => void,
+  onCompletion?: () => void
+): Unsubscribe;
+export function onSnapshot<T>(
+  reference: firestore.DocumentReference<T>,
+  options: firestore.SnapshotListenOptions,
+  onNext: (snapshot: firestore.DocumentSnapshot<T>) => void,
+  onError?: (error: firestore.FirestoreError) => void,
+  onCompletion?: () => void
+): Unsubscribe;
+export function onSnapshot<T>(
+  query: firestore.Query<T>,
+  observer: {
+    next?: (snapshot: firestore.QuerySnapshot<T>) => void;
+    error?: (error: firestore.FirestoreError) => void;
+    complete?: () => void;
+  }
+): Unsubscribe;
+export function onSnapshot<T>(
+  query: firestore.Query<T>,
+  options: firestore.SnapshotListenOptions,
+  observer: {
+    next?: (snapshot: firestore.QuerySnapshot<T>) => void;
+    error?: (error: firestore.FirestoreError) => void;
+    complete?: () => void;
+  }
+): Unsubscribe;
+export function onSnapshot<T>(
+  query: firestore.Query<T>,
+  onNext: (snapshot: firestore.QuerySnapshot<T>) => void,
+  onError?: (error: firestore.FirestoreError) => void,
+  onCompletion?: () => void
+): Unsubscribe;
+export function onSnapshot<T>(
+  query: firestore.Query<T>,
+  options: firestore.SnapshotListenOptions,
+  onNext: (snapshot: firestore.QuerySnapshot<T>) => void,
+  onError?: (error: firestore.FirestoreError) => void,
+  onCompletion?: () => void
+): Unsubscribe;
+export function onSnapshot<T>(
+  ref: firestore.Query<T> | firestore.DocumentReference<T>,
+  ...args: unknown[]
+): Unsubscribe {
+  let options: firestore.SnapshotListenOptions = {
+    includeMetadataChanges: false
+  };
+  let currArg = 0;
+  if (typeof args[currArg] === 'object' && !isPartialObserver(args[currArg])) {
+    options = args[currArg] as firestore.SnapshotListenOptions;
+    currArg++;
+  }
+
+  const internalOptions = {
+    includeMetadataChanges: options.includeMetadataChanges
+  };
+
+  if (isPartialObserver(args[currArg])) {
+    const userObserver = args[currArg] as PartialObserver<
+      firestore.QuerySnapshot<T>
+    >;
+    args[currArg] = userObserver.next?.bind(userObserver);
+    args[currArg + 1] = userObserver.error?.bind(userObserver);
+    args[currArg + 2] = userObserver.complete?.bind(userObserver);
+  }
+
+  let asyncObserver: Promise<Unsubscribe>;
+
+  if (ref instanceof DocumentReference) {
+    const firestore = cast(ref.firestore, Firestore);
+
+    const observer: PartialObserver<ViewSnapshot> = {
+      next: snapshot => {
+        if (args[currArg]) {
+          (args[currArg] as NextFn<firestore.DocumentSnapshot<T>>)(
+            convertToDocSnapshot(firestore, ref, snapshot)
+          );
+        }
+      },
+      error: args[currArg + 1] as ErrorFn,
+      complete: args[currArg + 2] as CompleteFn
+    };
+
+    asyncObserver = firestore
+      ._getFirestoreClient()
+      .then(firestoreClient =>
+        addDocSnapshotListener(
+          firestoreClient,
+          ref._key,
+          internalOptions,
+          observer
+        )
+      );
+  } else {
+    const query = cast<Query<T>>(ref, Query);
+    const firestore = cast(query, Firestore);
+
+    const observer: PartialObserver<ViewSnapshot> = {
+      next: snapshot => {
+        if (args[currArg]) {
+          (args[currArg] as NextFn<firestore.QuerySnapshot<T>>)(
+            new QuerySnapshot(firestore, query, snapshot)
+          );
+        }
+      },
+      error: args[currArg + 1] as ErrorFn,
+      complete: args[currArg + 2] as CompleteFn
+    };
+
+    validateHasExplicitOrderByForLimitToLast(query._query);
+
+    asyncObserver = firestore
+      ._getFirestoreClient()
+      .then(firestoreClient =>
+        addQuerySnapshotListener(
+          firestoreClient,
+          query._query,
+          internalOptions,
+          observer
+        )
+      );
+  }
+
+  // TODO(firestorexp): Add test that verifies that we don't raise a snapshot if
+  // unsubscribe is called before `asyncObserver` resolves.
+  return () => {
+    // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    asyncObserver.then(unsubscribe => unsubscribe());
+  };
+}
+
 /**
  * Converts a ViewSnapshot that contains the single document specified by `ref`
  * to a DocumentSnapshot.
diff --git a/packages/firestore/exp/src/api/snapshot.ts b/packages/firestore/exp/src/api/snapshot.ts
index 9e7ce22ad67..17a95bee6fd 100644
--- a/packages/firestore/exp/src/api/snapshot.ts
+++ b/packages/firestore/exp/src/api/snapshot.ts
@@ -121,15 +121,21 @@ export class QueryDocumentSnapshot<T = firestore.DocumentData>
 
 export class QuerySnapshot<T = firestore.DocumentData>
   implements firestore.QuerySnapshot<T> {
+  readonly metadata: SnapshotMetadata;
+
   private _cachedChanges?: Array<firestore.DocumentChange<T>>;
   private _cachedChangesIncludeMetadataChanges?: boolean;
 
   constructor(
     readonly _firestore: Firestore,
     readonly query: Query<T>,
-    readonly _snapshot: ViewSnapshot,
-    readonly metadata: SnapshotMetadata
-  ) {}
+    readonly _snapshot: ViewSnapshot
+  ) {
+    this.metadata = new SnapshotMetadata(
+      _snapshot.hasPendingWrites,
+      _snapshot.fromCache
+    );
+  }
 
   get docs(): Array<firestore.QueryDocumentSnapshot<T>> {
     const result: Array<firestore.QueryDocumentSnapshot<T>> = [];
@@ -154,7 +160,7 @@ export class QuerySnapshot<T = firestore.DocumentData>
         thisArg,
         this._convertToDocumentSnapshot(
           doc,
-          this.metadata.fromCache,
+          this._snapshot.fromCache,
           this._snapshot.mutatedKeys.has(doc.key)
         )
       );
diff --git a/packages/firestore/lite/src/api/reference.ts b/packages/firestore/lite/src/api/reference.ts
index 93d57408f91..845ab52faa4 100644
--- a/packages/firestore/lite/src/api/reference.ts
+++ b/packages/firestore/lite/src/api/reference.ts
@@ -48,7 +48,8 @@ import { hardAssert } from '../../../src/util/assert';
 import { DeleteMutation, Precondition } from '../../../src/model/mutation';
 import {
   applyFirestoreDataConverter,
-  BaseQuery
+  BaseQuery,
+  validateHasExplicitOrderByForLimitToLast
 } from '../../../src/api/database';
 import { FieldPath } from './field_path';
 import { cast } from './util';
@@ -417,6 +418,7 @@ export function getQuery<T>(
   query: firestore.Query<T>
 ): Promise<firestore.QuerySnapshot<T>> {
   const internalQuery = cast<Query<T>>(query, Query);
+  validateHasExplicitOrderByForLimitToLast(internalQuery._query);
   return internalQuery.firestore._getDatastore().then(async datastore => {
     const result = await invokeRunQueryRpc(datastore, internalQuery._query);
     const docs = result.map(
diff --git a/packages/firestore/src/api/database.ts b/packages/firestore/src/api/database.ts
index 12c0c73a870..13e7306a111 100644
--- a/packages/firestore/src/api/database.ts
+++ b/packages/firestore/src/api/database.ts
@@ -1270,7 +1270,7 @@ export class DocumentReference<T = firestore.DocumentData>
 }
 
 /** Registers an internal snapshot listener for `ref`. */
-function addDocSnapshotListener(
+export function addDocSnapshotListener(
   firestoreClient: FirestoreClient,
   key: DocumentKey,
   options: ListenOptions,
@@ -1698,17 +1698,6 @@ export class BaseQuery {
     return new Bound(components, before);
   }
 
-  protected validateHasExplicitOrderByForLimitToLast(
-    query: InternalQuery
-  ): void {
-    if (query.hasLimitToLast() && query.explicitOrderBy.length === 0) {
-      throw new FirestoreError(
-        Code.UNIMPLEMENTED,
-        'limitToLast() queries require specifying at least one orderBy() clause'
-      );
-    }
-  }
-
   /**
    * Parses the given documentIdValue into a ReferenceValue, throwing
    * appropriate errors if the value is anything other than a DocumentReference
@@ -1879,6 +1868,17 @@ export class BaseQuery {
   }
 }
 
+export function validateHasExplicitOrderByForLimitToLast(
+  query: InternalQuery
+): void {
+  if (query.hasLimitToLast() && query.explicitOrderBy.length === 0) {
+    throw new FirestoreError(
+      Code.UNIMPLEMENTED,
+      'limitToLast() queries require specifying at least one orderBy() clause'
+    );
+  }
+}
+
 export class Query<T = firestore.DocumentData> extends BaseQuery
   implements firestore.Query<T> {
   constructor(
@@ -2158,7 +2158,7 @@ export class Query<T = firestore.DocumentData> extends BaseQuery
       complete: args[currArg + 2] as CompleteFn
     };
 
-    this.validateHasExplicitOrderByForLimitToLast(this._query);
+    validateHasExplicitOrderByForLimitToLast(this._query);
     const firestoreClient = this.firestore.ensureClientConfigured();
     return addQuerySnapshotListener(
       firestoreClient,
@@ -2171,7 +2171,7 @@ export class Query<T = firestore.DocumentData> extends BaseQuery
   get(options?: firestore.GetOptions): Promise<firestore.QuerySnapshot<T>> {
     validateBetweenNumberOfArgs('Query.get', arguments, 0, 1);
     validateGetOptions('Query.get', options);
-    this.validateHasExplicitOrderByForLimitToLast(this._query);
+    validateHasExplicitOrderByForLimitToLast(this._query);
 
     const firestoreClient = this.firestore.ensureClientConfigured();
     return (options && options.source === 'cache'
@@ -2228,7 +2228,7 @@ export function getDocsViaSnapshotListener(
 }
 
 /** Registers an internal snapshot listener for `query`. */
-function addQuerySnapshotListener(
+export function addQuerySnapshotListener(
   firestore: FirestoreClient,
   query: InternalQuery,
   options: ListenOptions,