BufferedEvents Read Path Protections (#709)

* Discard and log persisted nil buffered events on read path
* Add nil safety checks to statsComputer

Author: shawnhathaway

common/persistence/executionStore.go

@@ -182,6 +182,12 @@ func (m *executionManagerImpl) DeserializeBufferedEvents(
 
 	events := make([]*historypb.HistoryEvent, 0)
 	for _, b := range blobs {
+		if b == nil {
+			// Should not happen, log and discard to prevent callers from consuming
+			m.logger.Warn("discarding nil buffered event")
+			continue
+		}
+
 		history, err := m.serializer.DeserializeBatchEvents(b)
 		if err != nil {
 			return nil, err

common/persistence/statsComputer.go

@@ -67,6 +67,11 @@ func (sc *statsComputer) computeMutableStateStats(req *InternalGetWorkflowExecut
 
 	for _, be := range req.State.BufferedEvents {
 		bufferedEventsCount++
+
+		if be == nil {
+			continue
+		}
+
 		bufferedEventsSize += len(be.Data)
 	}
 
@@ -182,6 +187,10 @@ func (sc *statsComputer) computeMutableStateUpdateStats(req *InternalUpdateWorkf
 }
 
 func computeExecutionInfoSize(executionInfo *InternalWorkflowExecutionInfo) int {
+	if executionInfo == nil {
+		return 0
+	}
+
 	size := len(executionInfo.WorkflowID)
 	size += len(executionInfo.TaskQueue)
 	size += len(executionInfo.WorkflowTypeName)
@@ -191,6 +200,10 @@ func computeExecutionInfoSize(executionInfo *InternalWorkflowExecutionInfo) int
 }
 
 func computeActivityInfoSize(ai *persistenceblobs.ActivityInfo) int {
+	if ai == nil {
+		return 0
+	}
+
 	size := len(ai.ActivityId)
 	if ai.ScheduledEvent != nil {
 		size += ai.ScheduledEvent.Size()
@@ -206,12 +219,19 @@ func computeActivityInfoSize(ai *persistenceblobs.ActivityInfo) int {
 }
 
 func computeTimerInfoSize(ti *persistenceblobs.TimerInfo) int {
+	if ti == nil {
+		return 0
+	}
+
 	size := len(ti.GetTimerId())
 
 	return size
 }
 
 func computeChildInfoSize(ci *persistenceblobs.ChildExecutionInfo) int {
+	if ci == nil {
+		return 0
+	}
 	size := 0
 	if ci.InitiatedEvent != nil {
 		size += ci.InitiatedEvent.Size()
@@ -223,6 +243,10 @@ func computeChildInfoSize(ci *persistenceblobs.ChildExecutionInfo) int {
 }
 
 func computeSignalInfoSize(si *persistenceblobs.SignalInfo) int {
+	if si == nil {
+		return 0
+	}
+
 	size := len(si.Name)
 	size += si.GetInput().Size()
 	size += len(si.Control)