fix(kbfs): don't crash on service notifications that arrive before KBFSOps is set

go/kbfs/libkbfs/favorites.go

@@ -70,6 +70,7 @@ type favReq struct {
 	favs               chan<- []favorites.Folder
 	favsAll            chan<- keybase1.FavoritesResult
 	homeTLFInfo        *homeTLFInfo
+	loadDisk           bool
 
 	// For asynchronous refreshes, pass in the Favorites from the server here
 	favResult *keybase1.FavoritesResult
@@ -130,6 +131,7 @@ type Favorites struct {
 	shutdownChan chan struct{}
 	muShutdown   sync.RWMutex
 	shutdown     bool
+	loopOnce     sync.Once
 }
 
 func newFavoritesWithChan(config Config, reqChan chan *favReq) *Favorites {
@@ -157,10 +159,19 @@ func newFavoritesWithChan(config Config, reqChan chan *favReq) *Favorites {
 		bufferedInterval: defaultFavoritesBufferedReqInterval,
 		shutdownChan:     make(chan struct{}),
 	}
-
 	return f
 }
 
+func (f *Favorites) startLoop() {
+	if f.disabled {
+		return
+	}
+	f.loopOnce.Do(func() {
+		f.loopWG.Add(1)
+		go f.loop()
+	})
+}
+
 // NewFavorites constructs a new Favorites instance.
 func NewFavorites(config Config) *Favorites {
 	return newFavoritesWithChan(config, make(chan *favReq, 100))
@@ -290,27 +301,25 @@ func (f *Favorites) writeCacheToDisk(ctx context.Context) error {
 // InitForTest starts the Favorites cache's internal processing loop without
 // loading cached favorites from disk.
 func (f *Favorites) InitForTest() {
-	if f.disabled {
-		return
-	}
-	go f.loop()
+	f.startLoop()
 }
 
-// Initialize loads the favorites cache from disk and starts listening for
-// requests asynchronously.
+// Initialize starts the processing loop and loads the favorites cache from
+// disk. Other methods start the loop on first use so notifications that
+// arrive before this can still be processed.
 func (f *Favorites) Initialize(ctx context.Context) {
 	if f.disabled {
 		return
 	}
-	// load cache from disk
-	err := f.readCacheFromDisk(ctx)
-	if err != nil {
+	req := &favReq{
+		ctx:      ctx,
+		loadDisk: true,
+		done:     make(chan struct{}),
+	}
+	if err := f.sendReq(ctx, req); err != nil {
 		f.log.CWarningf(
 			ctx, "Failed to read cached favorites from disk: %v", err)
 	}
-
-	// launch background loop
-	go f.loop()
 }
 
 func (f *Favorites) closeReq(req *favReq, err error) {
@@ -441,6 +450,10 @@ func (f *Favorites) handleReq(req *favReq) (err error) {
 		}
 	}()
 
+	if req.loadDisk {
+		return f.readCacheFromDisk(req.ctx)
+	}
+
 	if req.refresh && !req.buffered {
 		<-f.refreshWaiting
 	}
@@ -674,7 +687,6 @@ func (f *Favorites) handleReq(req *favReq) (err error) {
 }
 
 func (f *Favorites) loop() {
-	f.loopWG.Add(1)
 	defer f.loopWG.Done()
 	bufferedTicker := time.NewTicker(f.bufferedInterval)
 	defer bufferedTicker.Stop()
@@ -759,13 +771,20 @@ func (f *Favorites) waitOnReq(ctx context.Context,
 	}
 }
 
-func (f *Favorites) sendReq(ctx context.Context, req *favReq) error {
+func (f *Favorites) enqueue(ctx context.Context, req *favReq) error {
+	f.startLoop()
 	f.wg.Add(1)
 	select {
 	case f.reqChan <- req:
+		return nil
 	case <-ctx.Done():
 		f.wg.Done()
-		err := ctx.Err()
+		return ctx.Err()
+	}
+}
+
+func (f *Favorites) sendReq(ctx context.Context, req *favReq) error {
+	if err := f.enqueue(ctx, req); err != nil {
 		f.closeReq(req, err)
 		return err
 	}
@@ -835,14 +854,8 @@ func (f *Favorites) AddAsync(ctx context.Context, fav favorites.ToAdd) {
 	// if the original context is canceled.
 	req, doSend := f.startOrJoinAddReq(context.Background(), fav)
 	if doSend {
-		f.wg.Add(1)
-		select {
-		case f.reqChan <- req:
-		case <-ctx.Done():
-			f.wg.Done()
-			err := ctx.Err()
+		if err := f.enqueue(ctx, req); err != nil {
 			f.closeReq(req, err)
-			return
 		}
 	}
 }
@@ -914,36 +927,27 @@ func (f *Favorites) RefreshCache(ctx context.Context, mode FavoritesRefreshMode)
 		done:    make(chan struct{}),
 		ctx:     context.Background(),
 	}
-	f.wg.Add(1)
 
 	if mode == FavoritesRefreshModeBlocking {
 		favResult, err := f.config.KBPKI().FavoriteList(ctx)
 		if err != nil {
 			f.log.CDebugf(ctx, "Failed to refresh cached Favorites: %+v", err)
-			// Because the request will not make it to the main processing
-			// loop, mark it as done and clear the refresh channel here.
-			f.wg.Done()
 			<-f.refreshWaiting
 			return
 		}
 		req.favResult = &favResult
 	}
-	select {
-	case f.reqChan <- req:
-		go func() {
-			<-req.done
-			if req.err != nil {
-				f.log.CDebugf(ctx, "Failed to refresh cached Favorites ("+
-					"error in main loop): %+v", req.err)
-			}
-		}()
-	case <-ctx.Done():
-		// Because the request will not make it to the main processing
-		// loop, mark it as done and clear the refresh channel here.
-		f.wg.Done()
+	if err := f.enqueue(ctx, req); err != nil {
 		<-f.refreshWaiting
 		return
 	}
+	go func() {
+		<-req.done
+		if req.err != nil {
+			f.log.CDebugf(ctx, "Failed to refresh cached Favorites ("+
+				"error in main loop): %+v", req.err)
+		}
+	}()
 }
 
 // RefreshCacheWhenMTimeChanged refreshes the cached favorites, but
@@ -960,6 +964,7 @@ func (f *Favorites) RefreshCacheWhenMTimeChanged(
 	if f.disabled || f.shutdown {
 		return
 	}
+	f.startLoop()
 
 	req := &favReq{
 		refresh:   true,
@@ -1003,13 +1008,7 @@ func (f *Favorites) ClearCache(ctx context.Context) {
 		done:    make(chan struct{}),
 		ctx:     context.Background(),
 	}
-	f.wg.Add(1)
-	select {
-	case f.reqChan <- req:
-	case <-ctx.Done():
-		f.wg.Done()
-		return
-	}
+	_ = f.enqueue(ctx, req)
 }
 
 // GetFolderWithFavFlags returns the a FolderWithFavFlags for give folder, if found.
@@ -1086,13 +1085,7 @@ func (f *Favorites) setHomeTLFInfo(ctx context.Context, info homeTLFInfo) {
 		done:        make(chan struct{}),
 		ctx:         context.Background(),
 	}
-	f.wg.Add(1)
-	select {
-	case f.reqChan <- req:
-	case <-ctx.Done():
-		f.wg.Done()
-		return
-	}
+	_ = f.enqueue(ctx, req)
 }
 
 // GetAll returns the logged-in user's list of favorite, new, and ignored TLFs.

go/kbfs/libkbfs/init.go

@@ -797,40 +797,34 @@ func doInit(
 
 	kbfsLog := config.MakeLogger("")
 
-	// Initialize Keybase service connection. This needs to happen before
-	// KBPKI client.
-	if keybaseServiceCn == nil {
-		keybaseServiceCn = keybaseDaemon{}
-	}
-	service, err := keybaseServiceCn.NewKeybaseService(
-		config, params, kbCtx, kbfsLog)
-	if err != nil {
-		return nil, fmt.Errorf("problem creating service: %s", err)
-	}
-	if registry := config.MetricsRegistry(); registry != nil {
-		service = NewKeybaseServiceMeasured(service, registry)
-	}
-	config.SetKeybaseService(service)
-
-	// Initialize KBPKI client (needed for KBFSOps, MD Server, and Chat).
+	// Initialize KBPKI client (needed for KBFSOps, MD Server, and Chat). It
+	// reaches the service through config, so it doesn't need it yet.
 	k := NewKBPKIClient(config, kbfsLog)
 	config.SetKBPKI(k)
 
-	// Initialize Chat client (for file edit notifications).
-	chat, err := keybaseServiceCn.NewChat(config, params, kbCtx, kbfsLog)
-	if err != nil {
-		return nil, fmt.Errorf("problem creating chat: %s", err)
-	}
-	config.SetChat(chat)
-
+	// Set up KBFSOps and MDOps before the service connection. Creating the
+	// connection registers the KBFS handlers, and the service can call them
+	// right away. None of these use the service until they're called.
 	initDoneCh := make(chan struct{})
 	kbfsOps := NewKBFSOpsStandard(kbCtx, config, initDoneCh)
-	defer close(initDoneCh)
+	// Handlers on the service connection wait for init (see
+	// waitForKBFSReady), so tell them how it ended.
+	initSucceeded := false
+	defer func() {
+		if initSucceeded {
+			close(initDoneCh)
+		} else {
+			kbfsOps.initFailed()
+		}
+	}()
 	config.SetKBFSOps(kbfsOps)
 	config.SetNotifier(kbfsOps)
 	config.SetKeyManager(NewKeyManagerStandard(config))
 	config.SetMDOps(NewMDOpsStandard(config))
 
+	// Also before the service connection: a login it delivers can create the
+	// disk block cache, which expects the disk limiter. The limiter only reads
+	// local config.
 	config.SetDiskBlockCacheFraction(getCacheFrac(
 		ctx, kbCtx, params.DiskBlockCacheFraction,
 		defaultDiskBlockCacheFraction, configBlockCacheDiskMaxFracStr, log))
@@ -843,6 +837,27 @@ func doInit(
 		return nil, err
 	}
 
+	// Initialize Keybase service connection.
+	if keybaseServiceCn == nil {
+		keybaseServiceCn = keybaseDaemon{}
+	}
+	service, err := keybaseServiceCn.NewKeybaseService(
+		config, params, kbCtx, kbfsLog)
+	if err != nil {
+		return nil, fmt.Errorf("problem creating service: %s", err)
+	}
+	if registry := config.MetricsRegistry(); registry != nil {
+		service = NewKeybaseServiceMeasured(service, registry)
+	}
+	config.SetKeybaseService(service)
+
+	// Initialize Chat client (for file edit notifications).
+	chat, err := keybaseServiceCn.NewChat(config, params, kbCtx, kbfsLog)
+	if err != nil {
+		return nil, fmt.Errorf("problem creating chat: %s", err)
+	}
+	config.SetChat(chat)
+
 	kbfsOps.favs.Initialize(ctx)
 
 	config.SetReporter(NewReporterKBPKI(config, 10, 1000))
@@ -969,6 +984,19 @@ func doInit(
 		}
 	}
 
+	if params.BGFlushDirOpBatchSize < 1 {
+		return nil, fmt.Errorf(
+			"Illegal sync batch size: %d", params.BGFlushDirOpBatchSize)
+	}
+	log.CDebugf(ctx, "Enabling a dir op batch size of %d",
+		params.BGFlushDirOpBatchSize)
+	config.SetBGFlushDirOpBatchSize(params.BGFlushDirOpBatchSize)
+
+	// Requests on the service connection have what they need from here on.
+	// Don't hold them for journaling, which can take a while. Nothing after
+	// this point may fail init, since released requests can't be recalled.
+	kbfsOps.initReady()
+
 	ctx60s, cancel := context.WithTimeout(ctx, 60*time.Second)
 	defer cancel()
 	// TODO: Don't turn on journaling if either -bserver or
@@ -983,18 +1011,11 @@ func doInit(
 		log.CDebugf(ctx, "Journaling enabled")
 	}
 
-	if params.BGFlushDirOpBatchSize < 1 {
-		return nil, fmt.Errorf(
-			"Illegal sync batch size: %d", params.BGFlushDirOpBatchSize)
-	}
-	log.CDebugf(ctx, "Enabling a dir op batch size of %d",
-		params.BGFlushDirOpBatchSize)
-	config.SetBGFlushDirOpBatchSize(params.BGFlushDirOpBatchSize)
-
 	if config.Mode().OldStorageRootCleaningEnabled() {
 		go cleanOldTempStorageRoots(config)
 	}
 
+	initSucceeded = true
 	return config, nil
 }