diff --git a/services/syncbase/server/mojo_impl.go b/services/syncbase/server/mojo_impl.go
index 9c1b654..555b8fa 100644
--- a/services/syncbase/server/mojo_impl.go
+++ b/services/syncbase/server/mojo_impl.go
@@ -409,15 +409,9 @@
 		Continued:    vc.Continued,
 	}
 
-	// Block until client acks receiving and processing the previous change before sending more.
-	// This effectively creates a flow control mechanism.
-	// TODO(aghassemi): Consider sending more than a single change event before
-	// blocking until receiving ack.
-	ack, err := s.proxy.OnChange(mc)
-	if !ack && err != nil {
-		err = verror.NewErrInternal(s.ctx)
-	}
-	return err
+	// proxy.OnChange() blocks until the client acks the previous invocation, 
+	// thus providing flow control.
+	return s.proxy.OnChange(mc)
 }
 
 func (s *watchGlobStreamImpl) Recv(_ interface{}) error {
@@ -653,21 +647,12 @@
 	if !ok {
 		return verror.NewErrInternal(s.ctx)
 	}
-
-	// Block until client acks receiving and processing the previous change before sending more.
-	// This effectively creates a flow control mechanism.
-	// TODO(aghassemi): Consider sending more than a single KeyValue before
-	// blocking until receiving ack.
-	ack, err := s.proxy.OnKeyValue(mojom.KeyValue{
+	// proxy.OnKeyValue() blocks until the client acks the previous invocation,
+	// thus providing flow control.
+	return s.proxy.OnKeyValue(mojom.KeyValue{
 		Key:   kv.Key,
 		Value: kv.Value,
 	})
-
-	if !ack && err != nil {
-		err = verror.NewErrInternal(s.ctx)
-	}
-
-	return err
 }
 
 func (s *scanStreamImpl) Recv(_ interface{}) error {