Improve handling of vplayer stalls

Signed-off-by: Matt Lord <mattalord@gmail.com>

go/vt/vttablet/tabletmanager/vreplication/relaylog.go

@@ -17,6 +17,7 @@ limitations under the License.
 package vreplication
 
 import (
+ "errors"
  "io"
  "sync"
  "time"
@@ -26,6 +27,16 @@ import (
  binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+const (
+ // At what point should we consider writing to the relay log write stalled
+ // and return an error. This stall can happen if vplayer is stuck in a
+ // loop trying to process the previous relay log contents. This can happen
+ // e.g. if the queries it's executing are doing table scans and it thus
+ // cannot finish the transaction wrapping the contents before it gets
+ // termined due to crossing the query server transaction timeout.
+ sendTimeout = 5 * time.Minute
+)
+
 type relayLog struct {
  ctx context.Context
  maxItems int
@@ -69,11 +80,19 @@ func (rl *relayLog) Send(events []*binlogdatapb.VEvent) error {
  rl.mu.Lock()
  defer rl.mu.Unlock()
 
+ ctx, cancel := context.WithTimeout(rl.ctx, sendTimeout)
+ defer cancel()
  if err := rl.checkDone(); err != nil {
  return err
  }
  for rl.curSize > rl.maxSize || len(rl.items) >= rl.maxItems {
  rl.canAccept.Wait()
+ select {
+ case <-ctx.Done():
+ return errors.New("relay log write stalled")
+ default:
+ }
+ // Be sure that the vplayer contex is not done.
  if err := rl.checkDone(); err != nil {
  return err
  }

go/vt/vttablet/tabletmanager/vreplication/vplayer.go

@@ -31,6 +31,7 @@ import (
  "vitess.io/vitess/go/vt/binlog/binlogplayer"
  "vitess.io/vitess/go/vt/log"
  "vitess.io/vitess/go/vt/vttablet"
+ "vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
  "vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/throttlerapp"
 
  binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
@@ -497,8 +498,19 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
  return nil
  }
  }
+
+ // The parent context is a cancelable context. We create a new one here with a timeout
+ // to unblock the case where we're never able to complete applying the current vstream
+ // packet last read from the relay log.
+ iCtx, iCancel := context.WithTimeout(ctx, tabletenv.NewCurrentConfig().Oltp.TxTimeout)
+ defer iCancel()
  for i, events := range items {
  for j, event := range events {
+ select {
+ case <-iCtx.Done():
+ return errors.New("timed out replaying events in the relay log")
+ default:
+ }
  if event.Timestamp != 0 {
  vp.lastTimestampNs = event.Timestamp * 1e9
  vp.timeOffsetNs = time.Now().UnixNano() - event.CurrentTime
@@ -523,7 +535,7 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
  continue
  }
  }
- if err := vp.applyEvent(ctx, event, mustSave); err != nil {
+ if err := vp.applyEvent(iCtx, event, mustSave); err != nil {
  if err != io.EOF {
  vp.vr.stats.ErrorCounts.Add([]string{"Apply"}, 1)
  log.Errorf("Error applying event: %s", err.Error())

go/vt/vttablet/tabletserver/vstreamer/engine.go

@@ -234,7 +234,6 @@ func (vse *Engine) validateBinlogRowImage(ctx context.Context, db dbconfigs.Conn
 // Stream starts a new stream.
 // This streams events from the binary logs
 func (vse *Engine) Stream(ctx context.Context, startPos string, tablePKs []*binlogdatapb.TableLastPK, filter *binlogdatapb.Filter, throttlerApp throttlerapp.Name, send func([]*binlogdatapb.VEvent) error) error {
-
  if err := vse.validateBinlogRowImage(ctx, vse.se.GetDBConnector()); err != nil {
  return err
  }