Merge pull request #105 from ydb-platform/fix-race-on-hb-commit-if-hb-inordered

fix race in hb commit on case if hbs are inordered

Author: VPolka

internal/processor/processor.go

@@ -390,8 +390,15 @@ func (processor *Processor) EnqueueHb(ctx context.Context, hb types.HbData) erro
 				"enqueued just during our commit",
 				zap.Uint64("step", hb.Step),
 				zap.Uint64("tx_id", hb.TxId),
+				zap.Uint32("reader_id", hb.StreamId.ReaderId),
+				zap.Int64("partition_id", hb.StreamId.PartitionId),
 				zap.Uint64("our_step", lastPosition.Step),
 				zap.Uint64("our_tx_id", lastPosition.TxId))
+			err := hb.CommitTopic()
+			if err != nil {
+				errMsg := fmt.Sprintf("EnqueueHb: Unable to commit topic")
+				return types.ReturnError(ctx, err, errMsg)
+			}
 			return nil
 		}
 		return processor.hbTracker.AddHb(ctx, hb)
@@ -445,7 +452,11 @@ func (processor *Processor) EnqueueTx(ctx context.Context, tx types.TxData) erro
 				zap.Uint64("tx_id", tx.TxId),
 				zap.Uint64("our_step", lastPosition.Step),
 				zap.Uint64("our_tx_id", lastPosition.TxId))
-			return nil
+			err := tx.CommitTopic()
+			if err != nil {
+				errMsg := fmt.Sprintf("EnqueueTx: Unable to commit topic")
+				return types.ReturnError(ctx, err, errMsg)
+			}
 		}
 		processor.txQueue.PushTx(tx)
 		return nil