Method sync

flytestdlib/cache/in_memory_auto_refresh.go:312–390

There are w.parallelizm instances of this function running all the time, each one will: - Retrieve an item from the workqueue - For each batch of the keys, call syncCb, which tells us if the items have been updated -- If any has, then overwrite the item in the cache. What happens when the number of

(ctx context.Context)

// * Plugin asks for update on item 2 - cache evicts item 1, stores 2 and returns it unchanged
// * Sync loop updates item 2, repeat
func (w *InMemoryAutoRefresh) sync(ctx context.Context) (err error) {
	defer func() {
		var isErr bool
		rVal := recover()
		if rVal == nil {
			return
		}

		if err, isErr = rVal.(error); isErr {
			err = fmt.Errorf("worker panic'd and is shutting down. Error: %w with Stack: %v", err, string(debug.Stack()))
		} else {
			err = fmt.Errorf("worker panic'd and is shutting down. Panic value: %v with Stack: %v", rVal, string(debug.Stack()))
		}

		logger.Error(ctx, err)
	}()

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			batch, shutdown := w.workqueue.Get()
			if shutdown {
				logger.Debugf(ctx, "Shutting down worker")
				return nil
			}
			// Since we create batches every time we sync, we will just remove the item from the queue here
			// regardless of whether it succeeded the sync or not.
			w.workqueue.Forget(batch)
			w.workqueue.Done(batch)

			newBatch := make(Batch, 0, len(*batch))
			for _, b := range *batch {
				itemID := b.GetID()
				w.processing.Delete(itemID)
				item, ok := w.lruMap.Get(itemID)
				if !ok {
					logger.Debugf(ctx, "item with id [%v] not found in cache", itemID)
					continue
				}
				if item.(Item).IsTerminal() {
					logger.Debugf(ctx, "item with id [%v] is terminal", itemID)
					continue
				}
				newBatch = append(newBatch, b)
			}
			if len(newBatch) == 0 {
				continue
			}

			t := w.metrics.SyncLatency.Start()
			updatedBatch, err := w.syncCb(ctx, newBatch)

			if err != nil {
				w.metrics.SyncErrors.Inc()
				logger.Errorf(ctx, "failed to get latest copy of a batch. Error: %v", err)
				t.Stop()