// Redeliver all outstanding messages that have expired.
762874
func (s *StanServer) performAckExpirationRedelivery(sub *subState, isStartup bool) {
	// Sort our messages outstanding from acksPending, grab some state and unlock.
	sub.Lock()
	sortedPendingMsgs := makeSortedPendingMsgs(sub.acksPending)
	if len(sortedPendingMsgs) == 0 {
		sub.clearAckTimer()
		sub.Unlock()
		return
	}
	expTime := int64(sub.ackWait)
	subject := sub.subject
	qs := sub.qstate
	clientID := sub.ClientID
	subID := sub.ID
	if sub.ackTimer == nil {
		s.setupAckTimer(sub, sub.ackWait)
	}
	if qs == nil {
		// If the client has some failed heartbeats, ignore this request.
		if sub.hasFailedHB {
			// Reset the timer
			sub.ackTimer.Reset(sub.ackWait)
			sub.Unlock()
			if s.debug {
				s.log.Debugf("[Client:%s] Skipping redelivery to subid=%d due to missed client heartbeat", clientID, subID)
			}
			return
		}
	}
	sub.Unlock()

	c := s.channels.get(subject)
	if c == nil {
		s.log.Errorf("[Client:%s] Aborting redelivery to subid=%d for non existing channel %s", clientID, subID, subject)
		sub.Lock()
		sub.clearAckTimer()
		sub.Unlock()
		return
	}

	// In cluster mode we will always redeliver to the same queue member.
	// This is to avoid to have to replicated sent/ack when a message would
	// be redelivered (removed from one member to be sent to another member)
	isClustered := s.isClustered

	now := time.Now().UnixNano()
	// limit is now plus a buffer of 15ms to avoid repeated timer callbacks.
	limit := now + int64(15*time.Millisecond)

	var (
		pick               *subState
		sent               bool
		tracePrinted       bool
		foundWithZero      bool
		nextExpirationTime int64
	)

	// We will move through acksPending(sorted) and see what needs redelivery.
	for _, pm := range sortedPendingMsgs {
		m := s.getMsgForRedelivery(c, sub, pm.seq)
		if m == nil {
			continue
		}
		// If we found any pm.expire with 0 in the array (due to a server restart),
		// ensure that all have now an expiration set, then reschedule right away.
		if foundWithZero || pm.expire == 0 {
			foundWithZero = true
			if pm.expire == 0 {
				sub.Lock()
				// Is message still pending?
				if _, present := sub.acksPending[pm.seq]; present {
					sub.acksPending[pm.seq] = m.Timestamp + expTime
				}
				sub.Unlock()
			}
			continue
		}

		// If this message has not yet expired, reset timer for next callback
		if pm.expire > limit {
			nextExpirationTime = pm.expire
			if !tracePrinted && s.trace {
				tracePrinted = true
				s.log.Tracef("[Client:%s] Redelivery for subid=%d, skipping seq=%d", clientID, subID, m.Sequence)
			}
			break
		}

		// Flag as redelivered.
		m.Redelivered = true

		// Handle QueueSubscribers differently, since we will choose best subscriber
		// to redeliver to, not necessarily the same one.
		// However, on startup, resends only to member that had previously this message
		// otherwise this could cause a message to be redelivered to multiple members.
		if !isClustered && qs != nil && !isStartup {
			qs.Lock()
			pick, sent, _ = s.sendMsgToQueueGroup(qs, m, forceDelivery)
			qs.Unlock()
			if pick == nil {
				s.log.Errorf("[Client:%s] Unable to find queue subscriber for subid=%d", clientID, subID)
				break
			}
			// If the message is redelivered to a different queue subscriber,
			// we need to process an implicit ack for the original subscriber.
			// We do this only after confirmation that it was successfully added
			// as pending on the other queue subscriber.
			if pick != sub && sent {
				s.processAck(c, sub, m.Sequence, false)
			}
		} else {
			sub.Lock()
			s.sendMsgToSub(sub, m, forceDelivery)
			sub.Unlock()
		}
	}
	if foundWithZero {
		// Restart expiration now that ackPending map's expire values are properly
		// set. Note that messages may have been added/removed in the meantime.
		s.performAckExpirationRedelivery(sub, isStartup)
		return
	}

	// Adjust the timer
	sub.adjustAckTimer(nextExpirationTime)
}
70111
func (q *SQSDispatcher) extendMessageVisibilityTimeout(receiptHandle *string) (<-chan time.Time, error) {
	visibilityTimeout := int64(float64(q.VisibilityHeartbeat) / float64(time.Second))

	_, err := q.sqs.ChangeMessageVisibility(&sqs.ChangeMessageVisibilityInput{
		QueueUrl:          aws.String(q.QueueURL),
		ReceiptHandle:     receiptHandle,
		VisibilityTimeout: aws.Int64(visibilityTimeout),
	})
	if err != nil {
		return nil, err
	}

	return q.after(q.VisibilityHeartbeat / 2), nil
}
432922
func (s *Session) Expired() error {
	if time.Now().After(s.ExpiresAt) {
		return &Error{
			Code: EForbidden,
			Msg:  ErrSessionExpired,
		}
	}

	return nil
}
745366
func (c *Client) Renew(ack *dhcp4.Packet) (*dhcp4.Packet, error) {
	return c.SendAndReadOne(c.RequestPacket(ack))
}
586277
func renameExpired(preparedExpiration time.Duration) error {
	if err := pkgPod.WalkPods(getDataDir(), pkgPod.IncludePreparedDir, func(p *pkgPod.Pod) {
		st := &syscall.Stat_t{}
		pp := p.Path()
		if err := syscall.Lstat(pp, st); err != nil {
			if err != syscall.ENOENT {
				stderr.PrintE(fmt.Sprintf("unable to stat %q, ignoring", pp), err)
			}
			return
		}

		if expiration := time.Unix(st.Ctim.Unix()).Add(preparedExpiration); time.Now().After(expiration) {
			stderr.Printf("moving expired prepared pod %q to garbage", p.UUID)
			if err := p.ToGarbage(); err != nil && err != os.ErrNotExist {
				stderr.PrintE("rename error", err)
			}
		}
	}); err != nil {
		return err
	}
	return nil
}
1325262
func (f *Fetcher) rescheduleComplete(complete *time.Timer) {
	// Short circuit if no headers are fetched
	if len(f.fetched) == 0 {
		return
	}
	// Otherwise find the earliest expiring announcement
	earliest := time.Now()
	for _, announces := range f.fetched {
		if earliest.After(announces[0].time) {
			earliest = announces[0].time
		}
	}
	complete.Reset(gatherSlack - time.Since(earliest))
}
1022079
func (r *RedisStorer) ResetExpiry(key string) error {
	return r.client.Expire(key, r.maxAge).Err()
}
887175
func (p *RedisProvider) Regenerate(oldsid, sid string) (_ session.RawStore, err error) {
	poldsid := p.prefix + oldsid
	psid := p.prefix + sid

	if p.Exist(sid) {
		return nil, fmt.Errorf("new sid '%s' already exists", sid)
	} else if !p.Exist(oldsid) {
		// Make a fake old session.
		if err = p.c.SetEx(poldsid, p.duration, "").Err(); err != nil {
			return nil, err
		}
	}

	if err = p.c.Rename(poldsid, psid).Err(); err != nil {
		return nil, err
	}

	var kv map[interface{}]interface{}
	kvs, err := p.c.Get(psid).Result()
	if err != nil {
		return nil, err
	}

	if len(kvs) == 0 {
		kv = make(map[interface{}]interface{})
	} else {
		kv, err = session.DecodeGob([]byte(kvs))
		if err != nil {
			return nil, err
		}
	}

	return NewRedisStore(p.c, p.prefix, sid, p.duration, kv), nil
}
943346
func (f *Filter) Retrieve() (all []*ReceivedMessage) {
	f.mutex.Lock()
	defer f.mutex.Unlock()

	all = make([]*ReceivedMessage, 0, len(f.Messages))
	for _, msg := range f.Messages {
		all = append(all, msg)
	}

	f.Messages = make(map[common.Hash]*ReceivedMessage) // delete old messages
	return all
}
481088
func (msg *Message) CloneOriginal() *Message {
	if msg.orig == nil {
		msg.FreezeOriginal()
	}

	clone := *msg
	clone.data.payload = make([]byte, len(msg.orig.payload))
	copy(clone.data.payload, msg.orig.payload)

	if msg.orig.metadata != nil {
		clone.data.metadata = msg.orig.metadata.Clone()
	} else {
		clone.data.metadata = nil
	}

	clone.SetStreamID(msg.origStreamID)
	return &clone
}
543490
func (b *LRUCache) ExpireNow(now time.Time) int {
	b.lock.Lock()
	defer b.lock.Unlock()

	i := 0
	for {
		e := b.expiredEntry(now)
		if e == nil {
			break
		}
		b.removeEntry(e)
		i += 1
	}
	return i
}
599594
func (s *sender) retransmitTimerExpired() bool {
	// Check if the timer actually expired or if it's a spurious wake due
	// to a previously orphaned runtime timer.
	if !s.resendTimer.checkExpiration() {
		return true
	}

	s.ep.stack.Stats().TCP.Timeouts.Increment()

	// Give up if we've waited more than a minute since the last resend.
	if s.rto >= 60*time.Second {
		return false
	}

	// Set new timeout. The timer will be restarted by the call to sendData
	// below.
	s.rto *= 2

	if s.fr.active {
		// We were attempting fast recovery but were not successful.
		// Leave the state. We don't need to update ssthresh because it
		// has already been updated when entered fast-recovery.
		s.leaveFastRecovery()
	}

	// See: https://tools.ietf.org/html/rfc6582#section-3.2 Step 4.
	// We store the highest sequence number transmitted in cases where
	// we were not in fast recovery.
	s.fr.last = s.sndNxt - 1

	s.cc.HandleRTOExpired()

	// Mark the next segment to be sent as the first unacknowledged one and
	// start sending again. Set the number of outstanding packets to 0 so
	// that we'll be able to retransmit.
	//
	// We'll keep on transmitting (or retransmitting) as we get acks for
	// the data we transmit.
	s.outstanding = 0

	// Expunge all SACK information as per https://tools.ietf.org/html/rfc6675#section-5.1
	//
	//  In order to avoid memory deadlocks, the TCP receiver is allowed to
	//  discard data that has already been selectively acknowledged. As a
	//  result, [RFC2018] suggests that a TCP sender SHOULD expunge the SACK
	//  information gathered from a receiver upon a retransmission timeout
	//  (RTO) "since the timeout might indicate that the data receiver has
	//  reneged." Additionally, a TCP sender MUST "ignore prior SACK
	//  information in determining which data to retransmit."
	//
	// NOTE: We take the stricter interpretation and just expunge all
	// information as we lack more rigorous checks to validate if the SACK
	// information is usable after an RTO.
	s.ep.scoreboard.Reset()
	s.writeNext = s.writeList.Front()
	s.sendData()

	return true
}
1078434
func (vals *ValidatorSet) applyRemovals(deletes []*Validator) {

	existing := vals.Validators

	merged := make([]*Validator, len(existing)-len(deletes))
	i := 0

	// Loop over deletes until we removed all of them.
	for len(deletes) > 0 {
		if bytes.Equal(existing[0].Address, deletes[0].Address) {
			deletes = deletes[1:]
		} else { // Leave it in the resulting slice.
			merged[i] = existing[0]
			i++
		}
		existing = existing[1:]
	}

	// Add the elements which are left.
	for j := 0; j < len(existing); j++ {
		merged[i] = existing[j]
		i++
	}

	vals.Validators = merged[:i]
}
425773
func (m *MySQLStore) cleanup(interval time.Duration, quit <-chan struct{}, done chan<- struct{}) {
	ticker := time.NewTicker(interval)

	defer func() {
		ticker.Stop()
	}()

	for {
		select {
		case <-quit:
			// Handle the quit signal.
			done <- struct{}{}
			return
		case <-ticker.C:
			// Delete expired sessions on each tick.
			err := m.deleteExpired()
			if err != nil {
				log.Printf("sessions: mysqlstore: unable to delete expired sessions: %v", err)
			}
		}
	}
}
1189551
func (s *Sender) SendWithRetries(msg *Message, to string, retries int) (result *Result, err error) {
	if err := checkUnrecoverableErrors(s, to, nil, msg, retries); err != nil {
		return nil, err
	}
	attempt, backoff := 0, BackoffInitialDelay
	for {
		attempt++
		result, err = s.SendNoRetry(msg, to)
		// NOTE: partial success for a device group message is considered successful

		tryAgain := false
		if attempt <= retries {
			if result != nil && (result.Error == ErrorUnavailable || result.Error == ErrorInternalServerError) {
				tryAgain = true
			} else if err != nil {
				if httpErr, isHTTPErr := err.(httpError); isHTTPErr {
					tryAgain = httpErr.statusCode >= http.StatusInternalServerError && httpErr.statusCode < 600
				}
			}
		}

		if tryAgain {
			sleepTime := backoff/2 + rand.Intn(backoff)
			time.Sleep(time.Duration(sleepTime) * time.Millisecond)
			backoff = min(2*backoff, MaxBackoffDelay)
		} else {
			break
		}
	}
	return
}
1235470
func (m *ExpirationManager) LazyRevoke(ctx context.Context, leaseID string) error {
	defer metrics.MeasureSince([]string{"expire", "lazy-revoke"}, time.Now())

	// Load the entry
	le, err := m.loadEntry(ctx, leaseID)
	if err != nil {
		return err
	}

	// If there is no entry, nothing to revoke
	if le == nil {
		return nil
	}

	le.ExpireTime = time.Now()
	{
		m.pendingLock.Lock()
		if err := m.persistEntry(ctx, le); err != nil {
			m.pendingLock.Unlock()
			return err
		}

		m.updatePendingInternal(le, 0)
		m.pendingLock.Unlock()
	}

	return nil
}
117345
func ClearRetainedMessage(config *Config, topic string, timeout time.Duration) error {
	return PublishMessage(config, &packet.Message{
		Topic:   topic,
		Payload: nil,
		QOS:     0,
		Retain:  true,
	}, timeout)
}