// oplogReplay takes in a channel of batched operations and applys them using the
// supplied function.  Returns an error if the apply operation fails.
172097
func oplogReplay(batches <-chan []interface{}, applyOps func([]interface{}) error) error {
	for batch := range batches {
		if err := applyOps(batch); err != nil {
			return err
		}
	}
	return nil
}
1306384
func (c *Conn) opError(op string, err error) error {
	return opError(op, err, c.local, c.remote)
}
661522
func (r *raftFSM) Apply(l *raft.Log) interface{} {
	s := r.server
	op := &spb.RaftOperation{}
	if err := op.Unmarshal(l.Data); err != nil {
		panic(err)
	}
	switch op.OpType {
	case spb.RaftOperation_Publish:
		// Message replication.
		var (
			c       *channel
			err     error
			lastSeq uint64
		)
		for _, msg := range op.PublishBatch.Messages {
			// This is a batch for a given channel, so lookup channel once.
			if c == nil {
				c, err = s.lookupOrCreateChannel(msg.Subject)
				// That should not be the case, but if it happens,
				// just bail out.
				if err == ErrChanDelInProgress {
					return nil
				}
				lastSeq, err = c.store.Msgs.LastSequence()
			}
			if err == nil && lastSeq < msg.Sequence-1 {
				err = s.raft.fsm.restoreMsgsFromSnapshot(c, lastSeq+1, msg.Sequence-1)
			}
			if err == nil {
				_, err = c.store.Msgs.Store(msg)
			}
			if err != nil {
				return fmt.Errorf("failed to store replicated message %d on channel %s: %v",
					msg.Sequence, msg.Subject, err)
			}
		}
		return nil
	case spb.RaftOperation_Connect:
		// Client connection create replication.
		return s.processConnect(op.ClientConnect.Request, op.ClientConnect.Refresh)
	case spb.RaftOperation_Disconnect:
		// Client connection close replication.
		return s.closeClient(op.ClientDisconnect.ClientID)
	case spb.RaftOperation_Subscribe:
		// Subscription replication.
		sub, err := s.processSub(nil, op.Sub.Request, op.Sub.AckInbox, op.Sub.ID)
		return &replicatedSub{sub: sub, err: err}
	case spb.RaftOperation_RemoveSubscription:
		fallthrough
	case spb.RaftOperation_CloseSubscription:
		// Close/Unsub subscription replication.
		isSubClose := op.OpType == spb.RaftOperation_CloseSubscription
		s.closeMu.Lock()
		err := s.unsubscribe(op.Unsub, isSubClose)
		s.closeMu.Unlock()
		return err
	case spb.RaftOperation_SendAndAck:
		if !s.isLeader() {
			s.processReplicatedSendAndAck(op.SubSentAck)
		}
		return nil
	case spb.RaftOperation_DeleteChannel:
		s.processDeleteChannel(op.Channel)
		return nil
	default:
		panic(fmt.Sprintf("unknown op type %s", op.OpType))
	}
}
946017
func (f *FSM) Apply(log *raft.Log) interface{} {
	var command Command
	err := yaml.Unmarshal(log.Data, &command)
	if err != nil {
		return &response{err: errors.Trace(err)}
	}
	if err := command.Validate(); err != nil {
		return &response{err: errors.Trace(err)}
	}

	f.mu.Lock()
	defer f.mu.Unlock()

	switch command.Operation {
	case OperationClaim:
		return f.claim(command.LeaseKey(), command.Holder, command.Duration)
	case OperationExtend:
		return f.extend(command.LeaseKey(), command.Holder, command.Duration)
	case OperationPin:
		return f.pin(command.LeaseKey(), command.PinEntity)
	case OperationUnpin:
		return f.unpin(command.LeaseKey(), command.PinEntity)
	case OperationSetTime:
		return f.setTime(command.OldTime, command.NewTime)
	default:
		return &response{err: errors.NotValidf("operation %q", command.Operation)}
	}
}
1062286
func (cl *CommandList) Play(pos int) {
	if pos < 0 {
		cl.cmdQ.PushBack(&command{"play", nil, cmdNoReturn})
	} else {
		cl.cmdQ.PushBack(&command{fmt.Sprintf("play %d", pos), nil, cmdNoReturn})
	}
}
957204
func applyChannel(channel string) error {
	// We don't embed the channels code because we expect this will eventually be part of kubectl
	glog.Infof("checking channel: %q", channel)

	out, err := execChannels("apply", "channel", channel, "--v=4", "--yes")
	glog.V(4).Infof("apply channel output was: %v", out)
	return err
}
786664
func getApplyOpsFunc(session *mgo.Session, alwaysUpsert bool) func([]interface{}) error {
	return func(ops []interface{}) error {
		var result map[string]interface{}
		if err := session.Run(bson.D{{"applyOps", ops}, {"alwaysUpsert", alwaysUpsert}}, &result); err != nil {
			return err
		}
		// We have to inspect the response from session.Run to determine if the oplog operation
		// was applied correctly.
		resultsArray, ok := result["results"].([]interface{})
		if !ok {
			return fmt.Errorf("Failed to cast %v as []interfaces{}", result["results"])
		}
		for index, opResult := range resultsArray {
			boolResult, ok := opResult.(bool)
			if !ok {
				return fmt.Errorf("Failed to cast %v as bool", opResult)
			}
			if !boolResult {
				failedOp := ops[index].(map[string]interface{})
				return NewFailedOperationError(failedOp)
			}
		}
		numApplied, ok := result["applied"].(int)
		if !ok {
			return fmt.Errorf("Failed to cast applied %s as int", numApplied)
		}
		if numApplied != len(ops) {
			return fmt.Errorf("Operations applied %s does not match operations sent %s", numApplied, len(ops))
		}
		return nil
	}
}
921246
func ReplayPointer(xu *xgbutil.XUtil) {
	xproto.AllowEvents(xu.Conn(), xproto.AllowReplayPointer, 0)
}
130140
func (l *Listener) opError(op string, err error) error {
	// No remote address for a Listener.
	return opError(op, err, l.Addr(), nil)
}
677322
func (th *TlfHistory) FlushRevision(rev kbfsmd.Revision) {
	th.lock.Lock()
	defer th.lock.Unlock()
	if th.unflushed == nil {
		return
	}
	lastToKeep := len(th.unflushed.notifications) - 1
	for ; lastToKeep >= 0; lastToKeep-- {
		if th.unflushed.notifications[lastToKeep].Revision > rev {
			break
		}
	}
	if lastToKeep < len(th.unflushed.notifications)-1 {
		th.unflushed.notifications = th.unflushed.notifications[:lastToKeep+1]
		// Invalidate the cached results.
		th.computed = false
		th.cachedLoggedInUser = ""
	}
}
1095518
func LogCallback(c context.Context, opname string) Callback {
	return func(err error, delay time.Duration) {
		logging.Fields{
			logging.ErrorKey: err,
			"opname":         opname,
			"delay":          delay,
		}.Warningf(c, "operation failed transiently")
	}
}
476433
func (mr *MockCharmMockRecorder) LXDProfile() *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "LXDProfile", reflect.TypeOf((*MockCharm)(nil).LXDProfile))
}
1367251
func (g *Gnostic) performActions(message proto.Message) (err error) {
	// Optionally resolve internal references.
	if g.resolveReferences {
		if g.sourceFormat == SourceFormatOpenAPI2 {
			document := message.(*openapi_v2.Document)
			_, err = document.ResolveReferences(g.sourceName)
		} else if g.sourceFormat == SourceFormatOpenAPI3 {
			document := message.(*openapi_v3.Document)
			_, err = document.ResolveReferences(g.sourceName)
		}
		if err != nil {
			return err
		}
	}
	// Optionally write proto in binary format.
	if g.binaryOutputPath != "" {
		g.writeBinaryOutput(message)
	}
	// Optionally write proto in text format.
	if g.textOutputPath != "" {
		g.writeTextOutput(message)
	}
	// Optionally write document in yaml and/or json formats.
	if g.yamlOutputPath != "" || g.jsonOutputPath != "" {
		g.writeJSONYAMLOutput(message)
	}
	// Call all specified plugins.
	messages := make([]*plugins.Message, 0)
	for _, p := range g.pluginCalls {
		pluginMessages, err := p.perform(message, g.sourceFormat, g.sourceName, g.timePlugins)
		if err != nil {
			writeFile(g.errorOutputPath, g.errorBytes(err), g.sourceName, "errors")
			defer os.Exit(-1) // run all plugins, even when some have errors
		}
		messages = append(messages, pluginMessages...)
	}
	if g.messageOutputPath != "" {
		g.writeMessagesOutput(&plugins.Messages{Messages: messages})
	} else {
		// Print any messages from the plugins
		if len(messages) > 0 {
			for _, message := range messages {
				fmt.Printf("%+v\n", message)
			}
		}
	}
	return nil
}
1272255
func NewReplayXHRArgs(requestID RequestID) *ReplayXHRArgs {
	args := new(ReplayXHRArgs)
	args.RequestID = requestID
	return args
}
514837
func (d *DB) replayLogFile(ve *versionEdit, fs db.FileSystem, filename string) (maxSeqNum uint64, err error) {
	file, err := fs.Open(filename)
	if err != nil {
		return 0, err
	}
	defer file.Close()

	var (
		mem      *memdb.MemDB
		batchBuf = new(bytes.Buffer)
		ikey     = make(internalKey, 512)
		rr       = record.NewReader(file)
	)
	for {
		r, err := rr.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			return 0, err
		}
		_, err = io.Copy(batchBuf, r)
		if err != nil {
			return 0, err
		}

		if batchBuf.Len() < batchHeaderLen {
			return 0, fmt.Errorf("leveldb: corrupt log file %q", filename)
		}
		b := Batch{batchBuf.Bytes()}
		seqNum := b.seqNum()
		seqNum1 := seqNum + uint64(b.count())
		if maxSeqNum < seqNum1 {
			maxSeqNum = seqNum1
		}

		if mem == nil {
			mem = memdb.New(&d.icmpOpts)
		}

		t := b.iter()
		for ; seqNum != seqNum1; seqNum++ {
			kind, ukey, value, ok := t.next()
			if !ok {
				return 0, fmt.Errorf("leveldb: corrupt log file %q", filename)
			}
			// Convert seqNum, kind and key into an internalKey, and add that ikey/value
			// pair to mem.
			//
			// TODO: instead of copying to an intermediate buffer (ikey), is it worth
			// adding a SetTwoPartKey(db.TwoPartKey{key0, key1}, value, opts) method to
			// memdb.MemDB? What effect does that have on the db.Comparer interface?
			//
			// The C++ LevelDB code does not need an intermediate copy because its memdb
			// implementation is a private implementation detail, and copies each internal
			// key component from the Batch format straight to the skiplist buffer.
			//
			// Go's LevelDB considers the memdb functionality to be useful in its own
			// right, and so leveldb/memdb is a separate package that is usable without
			// having to import the top-level leveldb package. That extra abstraction
			// means that we need to copy to an intermediate buffer here, to reconstruct
			// the complete internal key to pass to the memdb.
			ikey = makeInternalKey(ikey, ukey, kind, seqNum)
			mem.Set(ikey, value, nil)
		}
		if len(t) != 0 {
			return 0, fmt.Errorf("leveldb: corrupt log file %q", filename)
		}

		// TODO: if mem is large enough, write it to a level-0 table and set mem = nil.

		batchBuf.Reset()
	}

	if mem != nil && !mem.Empty() {
		meta, err := d.writeLevel0Table(fs, mem)
		if err != nil {
			return 0, err
		}
		ve.newFiles = append(ve.newFiles, newFileEntry{level: 0, meta: meta})
		// Strictly speaking, it's too early to delete meta.fileNum from d.pendingOutputs,
		// but we are replaying the log file, which happens before Open returns, so there
		// is no possibility of deleteObsoleteFiles being called concurrently here.
		delete(d.pendingOutputs, meta.fileNum)
	}

	return maxSeqNum, nil
}
627746
func (pb *playback) replayConsoleLoop() int {
	for {
		fmt.Printf("> ")
		bufReader := bufio.NewReader(os.Stdin)
		line, more, err := bufReader.ReadLine()
		if more {
			cmn.Exit("input is too long")
		} else if err != nil {
			cmn.Exit(err.Error())
		}

		tokens := strings.Split(string(line), " ")
		if len(tokens) == 0 {
			continue
		}

		switch tokens[0] {
		case "next":
			// "next" -> replay next message
			// "next N" -> replay next N messages

			if len(tokens) == 1 {
				return 0
			}
			i, err := strconv.Atoi(tokens[1])
			if err != nil {
				fmt.Println("next takes an integer argument")
			} else {
				return i
			}

		case "back":
			// "back" -> go back one message
			// "back N" -> go back N messages

			// NOTE: "back" is not supported in the state machine design,
			// so we restart and replay up to

			ctx := context.Background()
			// ensure all new step events are regenerated as expected

			newStepSub, err := pb.cs.eventBus.Subscribe(ctx, subscriber, types.EventQueryNewRoundStep)
			if err != nil {
				cmn.Exit(fmt.Sprintf("failed to subscribe %s to %v", subscriber, types.EventQueryNewRoundStep))
			}
			defer pb.cs.eventBus.Unsubscribe(ctx, subscriber, types.EventQueryNewRoundStep)

			if len(tokens) == 1 {
				if err := pb.replayReset(1, newStepSub); err != nil {
					pb.cs.Logger.Error("Replay reset error", "err", err)
				}
			} else {
				i, err := strconv.Atoi(tokens[1])
				if err != nil {
					fmt.Println("back takes an integer argument")
				} else if i > pb.count {
					fmt.Printf("argument to back must not be larger than the current count (%d)\n", pb.count)
				} else {
					if err := pb.replayReset(i, newStepSub); err != nil {
						pb.cs.Logger.Error("Replay reset error", "err", err)
					}
				}
			}

		case "rs":
			// "rs" -> print entire round state
			// "rs short" -> print height/round/step
			// "rs <field>" -> print another field of the round state

			rs := pb.cs.RoundState
			if len(tokens) == 1 {
				fmt.Println(rs)
			} else {
				switch tokens[1] {
				case "short":
					fmt.Printf("%v/%v/%v\n", rs.Height, rs.Round, rs.Step)
				case "validators":
					fmt.Println(rs.Validators)
				case "proposal":
					fmt.Println(rs.Proposal)
				case "proposal_block":
					fmt.Printf("%v %v\n", rs.ProposalBlockParts.StringShort(), rs.ProposalBlock.StringShort())
				case "locked_round":
					fmt.Println(rs.LockedRound)
				case "locked_block":
					fmt.Printf("%v %v\n", rs.LockedBlockParts.StringShort(), rs.LockedBlock.StringShort())
				case "votes":
					fmt.Println(rs.Votes.StringIndented("  "))

				default:
					fmt.Println("Unknown option", tokens[1])
				}
			}
		case "n":
			fmt.Println(pb.count)
		}
	}
}
853450
func (e *Engine) replayWAL() error {
	if !e.config.WAL.Enabled {
		return nil
	}
	now := time.Now()

	walPaths, err := wal.SegmentFileNames(e.wal.Path())
	if err != nil {
		return err
	}

	// TODO(jeff): we should just do snapshots and wait for them so that we don't hit
	// OOM situations when reloading huge WALs.

	// Disable the max size during loading
	limit := e.engine.Cache.MaxSize()
	defer func() { e.engine.Cache.SetMaxSize(limit) }()
	e.engine.Cache.SetMaxSize(0)

	// Execute all the entries in the WAL again
	reader := wal.NewWALReader(walPaths)
	reader.WithLogger(e.logger)
	err = reader.Read(func(entry wal.WALEntry) error {
		switch en := entry.(type) {
		case *wal.WriteWALEntry:
			points := tsm1.ValuesToPoints(en.Values)
			err := e.writePointsLocked(context.Background(), tsdb.NewSeriesCollection(points), en.Values)
			if _, ok := err.(tsdb.PartialWriteError); ok {
				err = nil
			}
			return err

		case *wal.DeleteBucketRangeWALEntry:
			return e.deleteBucketRangeLocked(en.OrgID, en.BucketID, en.Min, en.Max)
		}

		return nil
	})

	e.logger.Info("Reloaded WAL",
		zap.String("path", e.wal.Path()),
		zap.Duration("duration", time.Since(now)),
		zap.Error(err))

	return err
}