// doChunkedDowngrades sends batched archive or delete messages to the
// block server for the given block pointers.  For deletes, it returns
// a list of block IDs that no longer have any references.
971145
func (fbm *folderBlockManager) doChunkedDowngrades(ctx context.Context,
	tlfID tlf.ID, ptrs []data.BlockPointer, archive bool) (
	[]kbfsblock.ID, error) {
	fbm.log.CDebugf(ctx, "Downgrading %d pointers (archive=%t)",
		len(ptrs), archive)
	bops := fbm.config.BlockOps()

	// Round up to find the number of chunks.
	numChunks := (len(ptrs) + numPointersToDowngradePerChunk - 1) /
		numPointersToDowngradePerChunk
	numWorkers := numChunks
	if numWorkers > maxParallelBlockPuts {
		numWorkers = maxParallelBlockPuts
	}
	chunks := make(chan []data.BlockPointer, numChunks)

	var wg sync.WaitGroup
	defer wg.Wait()

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	type workerResult struct {
		zeroRefCounts []kbfsblock.ID
		err           error
	}

	chunkResults := make(chan workerResult, numChunks)
	worker := func() {
		defer wg.Done()
		for chunk := range chunks {
			var res workerResult
			fbm.log.CDebugf(ctx, "Downgrading chunk of %d pointers", len(chunk))
			if archive {
				res.err = bops.Archive(ctx, tlfID, chunk)
			} else {
				var liveCounts map[kbfsblock.ID]int
				liveCounts, res.err = bops.Delete(ctx, tlfID, chunk)
				if res.err == nil {
					for id, count := range liveCounts {
						if count == 0 {
							res.zeroRefCounts = append(res.zeroRefCounts, id)
						}
					}
				}
			}
			chunkResults <- res
			select {
			// return early if the context has been canceled
			case <-ctx.Done():
				return
			default:
			}
		}
	}
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go worker()
	}

	for start := 0; start < len(ptrs); start += numPointersToDowngradePerChunk {
		end := start + numPointersToDowngradePerChunk
		if end > len(ptrs) {
			end = len(ptrs)
		}
		chunks <- ptrs[start:end]
	}
	close(chunks)

	var zeroRefCounts []kbfsblock.ID
	for i := 0; i < numChunks; i++ {
		result := <-chunkResults
		if result.err != nil {
			// deferred cancel will stop the other workers.
			return nil, result.err
		}
		zeroRefCounts = append(zeroRefCounts, result.zeroRefCounts...)
	}
	return zeroRefCounts, nil
}
904373
func (fbm *folderBlockManager) runUnlessShutdown(
	fn func(ctx context.Context) error) error {
	ctx := fbm.ctxWithFBMID(context.Background())
	return fbm.runUnlessShutdownWithCtx(ctx, fn)
}
1151336
func getSortedDeletePairs(deleteMap BatchDeleteMap) batchDeletePairList {
	res := batchDeletePairList{}
	for item, updates := range deleteMap {
		res = append(res, batchDeletePair{item, updates})
	}

	sort.Sort(res)
	return res
}
874792
func (b *BlockServerRemote) ArchiveBlockReferences(ctx context.Context,
	tlfID tlf.ID, contexts kbfsblock.ContextMap) (err error) {
	ctx = rpc.WithFireNow(ctx)
	b.log.LazyTrace(ctx, "BServer: ArchiveRef %v", contexts)
	defer func() {
		b.log.LazyTrace(ctx, "BServer: ArchiveRef %v done (err=%v)", contexts, err)
		if err != nil {
			b.deferLog.CWarningf(ctx, "ArchiveBlockReferences batch size=%d err=%v", len(contexts), err)
		} else {
			b.deferLog.CDebugf(ctx, "ArchiveBlockReferences batch size=%d", len(contexts))
		}
	}()
	_, err = kbfsblock.BatchDowngradeReferences(ctx, b.log, tlfID, contexts, true, b.putConn.getClient())
	return err
}
630844
func dbRemoveSpendJournalEntry(dbTx database.Tx, blockHash *chainhash.Hash) error {
	spendBucket := dbTx.Metadata().Bucket(spendJournalBucketName)
	return spendBucket.Delete(blockHash[:])
}
725586
func (pgb *ChainDB) PurgeBestBlocks(N int64) (*dbtypes.DeletionSummary, int64, error) {
	res, height, _, err := DeleteBlocks(pgb.ctx, N, pgb.db)
	if err != nil {
		return nil, height, pgb.replaceCancelError(err)
	}

	summary := dbtypes.DeletionSummarySlice(res).Reduce()

	return &summary, height, err
}
725110
func (p *peer) AsyncSendNewBlock(block *types.Block, td *big.Int) {
	select {
	case p.queuedProps <- &propEvent{block: block, td: td}:
		p.knownBlocks.Add(block.Hash())
	default:
		p.Log().Debug("Dropping block propagation", "number", block.NumberU64(), "hash", block.Hash())
	}
}
1365643
func (router *Router) Downgraded() bool {
	return router.mainHandler != nil && router.requestHandler == nil
}
44789
func normalizeBlocks(a []Block) []Block {
	a = joinAdjacentTextBlocks(a)
	a = trimTrailingEmptyTextBlocks(a)
	return a
}
671290
func (t *requestTracker) markedDown() error {
	ss := atomic.LoadInt32(&t.shutdownState)
	if ss != running {
		reason := t.shutdownReason.Load().(*error)
		if reason == nil {
			return nil
		}
		return *reason
	}
	return nil
}
1352810
func (df *DirtyFile) ResetSyncingBlocksToDirty() {
	df.lock.Lock()
	defer df.lock.Unlock()
	// Reset all syncing blocks to just be dirty again
	syncFinishedNeeded := false
	for ptr, state := range df.fileBlockStates {
		if state.orphaned {
			// This block will never be sync'd again, so clear any
			// bytes from the buffer.
			if state.sync == blockSyncing {
				df.dirtyBcache.UpdateUnsyncedBytes(df.Path.Tlf,
					-state.syncSize, true)
			} else if state.sync == blockSynced {
				// Some blocks did finish, so we might be able to
				// increase our buffer size.
				syncFinishedNeeded = true
			}
			state.syncSize = 0
			delete(df.fileBlockStates, ptr)
			continue
		}
		if state.sync == blockSynced {
			// Re-dirty the unsynced bytes (but don't touch the total
			// bytes).
			df.dirtyBcache.BlockSyncFinished(df.Path.Tlf, -state.syncSize)
		} else if state.sync == blockSyncing {
			df.dirtyBcache.UpdateSyncingBytes(df.Path.Tlf, -state.syncSize)
		}
		if state.sync != blockNotSyncing {
			state.copy = blockAlreadyCopied
			state.sync = blockNotSyncing
			state.syncSize = 0
			df.fileBlockStates[ptr] = state
		}
	}
	if syncFinishedNeeded {
		df.dirtyBcache.SyncFinished(df.Path.Tlf, df.totalSyncBytes)
	}
	df.totalSyncBytes = 0 // all the blocks need to be re-synced.
}
226361
func (c DiskBlockCacheClient) DeleteBlocks(ctx context.Context, blockIDs [][]byte) (res DeleteBlocksRes, err error) {
	__arg := DeleteBlocksArg{BlockIDs: blockIDs}
	err = c.Cli.Call(ctx, "kbgitkbfs.1.DiskBlockCache.DeleteBlocks", []interface{}{__arg}, &res)
	return
}
145090
func (db *DB) DeleteBlock(blockhash string) (NSummaryRows, NStakeInfoRows int64, err error) {
	// Attempt to purge the block data.
	NSummaryRows, NStakeInfoRows, err = db.deleteBlock(blockhash)
	if err != nil {
		return
	}

	err = db.updateHeights()
	return
}
407095
func (pgb *ChainDB) DeleteDuplicates(barLoad chan *dbtypes.ProgressBarLoad) error {
	allDuplicates := []dropDuplicatesInfo{
		// Remove duplicate vins
		{TableName: "vins", DropDupsFunc: pgb.DeleteDuplicateVins},

		// Remove duplicate vouts
		{TableName: "vouts", DropDupsFunc: pgb.DeleteDuplicateVouts},

		// Remove duplicate transactions
		{TableName: "transactions", DropDupsFunc: pgb.DeleteDuplicateTxns},

		// Remove duplicate agendas
		{TableName: "agendas", DropDupsFunc: pgb.DeleteDuplicateAgendas},

		// Remove duplicate agenda_votes
		{TableName: "agenda_votes", DropDupsFunc: pgb.DeleteDuplicateAgendaVotes},
	}

	var err error
	for _, val := range allDuplicates {
		msg := fmt.Sprintf("Finding and removing duplicate %s entries...", val.TableName)
		if barLoad != nil {
			barLoad <- &dbtypes.ProgressBarLoad{BarID: dbtypes.InitialDBLoad, Subtitle: msg}
		}
		log.Info(msg)

		var numRemoved int64
		if numRemoved, err = val.DropDupsFunc(); err != nil {
			return fmt.Errorf("delete %s duplicate failed: %v", val.TableName, err)
		}

		msg = fmt.Sprintf("Removed %d duplicate %s entries.", numRemoved, val.TableName)
		if barLoad != nil {
			barLoad <- &dbtypes.ProgressBarLoad{BarID: dbtypes.InitialDBLoad, Subtitle: msg}
		}
		log.Info(msg)
	}
	// Signal task is done
	if barLoad != nil {
		barLoad <- &dbtypes.ProgressBarLoad{BarID: dbtypes.InitialDBLoad, Subtitle: " "}
	}
	return nil
}
1069153
func (c *Core) scheduleUpgradeCleanup(ctx context.Context) error {
	// List the upgrades
	upgrades, err := c.barrier.List(ctx, keyringUpgradePrefix)
	if err != nil {
		return errwrap.Wrapf("failed to list upgrades: {{err}}", err)
	}

	// Nothing to do if no upgrades
	if len(upgrades) == 0 {
		return nil
	}

	// Schedule cleanup for all of them
	time.AfterFunc(keyRotateGracePeriod, func() {
		sealed, err := c.barrier.Sealed()
		if err != nil {
			c.logger.Warn("failed to check barrier status at upgrade cleanup time")
			return
		}
		if sealed {
			c.logger.Warn("barrier sealed at upgrade cleanup time")
			return
		}
		for _, upgrade := range upgrades {
			path := fmt.Sprintf("%s%s", keyringUpgradePrefix, upgrade)
			if err := c.barrier.Delete(ctx, path); err != nil {
				c.logger.Error("failed to cleanup upgrade", "path", path, "error", err)
			}
		}
	})
	return nil
}
726343
func (dbclient *CouchDatabase) BatchUpdateDocuments(documents []*CouchDoc) ([]*BatchUpdateResponse, error) {
	dbName := dbclient.DBName

	if logger.IsEnabledFor(zapcore.DebugLevel) {
		documentIdsString, err := printDocumentIds(documents)
		if err == nil {
			logger.Debugf("[%s] Entering BatchUpdateDocuments()  document ids=[%s]", dbName, documentIdsString)
		} else {
			logger.Debugf("[%s] Entering BatchUpdateDocuments()  Could not print document ids due to error: %+v", dbName, err)
		}
	}

	batchUpdateURL, err := url.Parse(dbclient.CouchInstance.conf.URL)
	if err != nil {
		logger.Errorf("URL parse error: %s", err)
		return nil, errors.Wrapf(err, "error parsing CouchDB URL: %s", dbclient.CouchInstance.conf.URL)
	}

	documentMap := make(map[string]interface{})

	var jsonDocumentMap []interface{}

	for _, jsonDocument := range documents {

		//create a document map
		var document = make(map[string]interface{})

		//unmarshal the JSON component of the CouchDoc into the document
		err = json.Unmarshal(jsonDocument.JSONValue, &document)
		if err != nil {
			return nil, errors.Wrap(err, "error unmarshalling json data")
		}

		//iterate through any attachments
		if len(jsonDocument.Attachments) > 0 {

			//create a file attachment map
			fileAttachment := make(map[string]interface{})

			//for each attachment, create a Base64Attachment, name the attachment,
			//add the content type and base64 encode the attachment
			for _, attachment := range jsonDocument.Attachments {
				fileAttachment[attachment.Name] = Base64Attachment{attachment.ContentType,
					base64.StdEncoding.EncodeToString(attachment.AttachmentBytes)}
			}

			//add attachments to the document
			document["_attachments"] = fileAttachment

		}

		//Append the document to the map of documents
		jsonDocumentMap = append(jsonDocumentMap, document)

	}

	//Add the documents to the "docs" item
	documentMap["docs"] = jsonDocumentMap

	bulkDocsJSON, err := json.Marshal(documentMap)
	if err != nil {
		return nil, errors.Wrap(err, "error marshalling json data")
	}

	//get the number of retries
	maxRetries := dbclient.CouchInstance.conf.MaxRetries

	resp, _, err := dbclient.handleRequest(http.MethodPost, "BatchUpdateDocuments", batchUpdateURL, bulkDocsJSON, "", "", maxRetries, true, nil, "_bulk_docs")
	if err != nil {
		return nil, err
	}
	defer closeResponseBody(resp)

	if logger.IsEnabledFor(zapcore.DebugLevel) {
		dump, _ := httputil.DumpResponse(resp, false)
		// compact debug log by replacing carriage return / line feed with dashes to separate http headers
		logger.Debugf("[%s] HTTP Response: %s", dbclient.DBName, bytes.Replace(dump, []byte{0x0d, 0x0a}, []byte{0x20, 0x7c, 0x20}, -1))
	}

	//handle as JSON document
	jsonResponseRaw, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, errors.Wrap(err, "error reading response body")
	}

	var jsonResponse = []*BatchUpdateResponse{}
	err2 := json.Unmarshal(jsonResponseRaw, &jsonResponse)
	if err2 != nil {
		return nil, errors.Wrap(err2, "error unmarshalling json data")
	}

	logger.Debugf("[%s] Exiting BatchUpdateDocuments() _bulk_docs response=[%s]", dbclient.DBName, string(jsonResponseRaw))

	return jsonResponse, nil

}
1221932
func (s *store) ResetLastUpdatedOldBlocksList() error {
	batch := leveldbhelper.NewUpdateBatch()
	batch.Delete(lastUpdatedOldBlocksKey)
	if err := s.db.WriteBatch(batch, true); err != nil {
		return err
	}
	s.isLastUpdatedOldBlocksSet = false
	return nil
}