fix(desktop): bound Slack Redux node decoder with context

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Fixes
 
 - Stop Slack channel, history, thread, and DM pagination with a clear error when a cursor repeats instead of looping indefinitely. Thanks @SebTardif! (#159)
+- Cancel hung Slack Desktop Node decoders when stopping sync, watch, or doctor. Thanks @SebTardif! (#163)
 
 ### Maintenance
 

docs/desktop-mode.md

@@ -79,6 +79,8 @@ This will:
 2. parse Local Storage and IndexedDB data
 3. merge supported rows into SQLite
 
+Ctrl-C cancels desktop sync, watch, and doctor, including a Node decoder that stops responding.
+
 ## Continuous Desktop Refresh
 
 Use `watch` to keep refreshing the DB from local desktop state:

internal/cli/app.go

@@ -318,7 +318,7 @@ func (a *App) runDoctor(ctx context.Context, configPath string, args []string, f
 	}
 	desktop := slackdesktop.Source{Path: cfg.Slack.Desktop.Path, Available: false}
 	if cfg.Slack.Desktop.Enabled {
-		desktop, err = slackdesktop.Inspect(cfg.Slack.Desktop.Path)
+		desktop, err = slackdesktop.Inspect(ctx, cfg.Slack.Desktop.Path)
 		if err != nil {
 			return err
 		}

internal/slackdesktop/desktop.go

@@ -254,7 +254,7 @@ func Discover(path string) (Source, error) {
 	return source, nil
 }
 
-func Inspect(path string) (Source, error) {
+func Inspect(ctx context.Context, path string) (Source, error) {
 	source, err := Discover(path)
 	if err != nil {
 		return Source{}, err
@@ -269,7 +269,7 @@ func Inspect(path string) (Source, error) {
 	}
 	defer func() { _ = os.RemoveAll(filepath.Dir(snapshot.Root)) }()
 
-	extracted, err := Extract(snapshot.Root)
+	extracted, err := Extract(ctx, snapshot.Root)
 	if err != nil {
 		return Source{}, err
 	}
@@ -329,7 +329,7 @@ func SnapshotPath(path string) (snapshot Snapshot, err error) {
 	return Snapshot{Root: target}, nil
 }
 
-func Extract(path string) (ExtractedData, error) {
+func Extract(ctx context.Context, path string) (ExtractedData, error) {
 	root, err := LoadRootState(filepath.Join(path, rootStateFile))
 	if err != nil && !os.IsNotExist(err) {
 		return ExtractedData{}, err
@@ -344,7 +344,7 @@ func Extract(path string) (ExtractedData, error) {
 	if err != nil && !os.IsNotExist(err) {
 		return ExtractedData{}, err
 	}
-	reduxStates, decodeSummary, err := extractIndexedDBStates(path)
+	reduxStates, decodeSummary, err := extractIndexedDBStates(ctx, path)
 	if err != nil {
 		return ExtractedData{}, err
 	}
@@ -386,7 +386,7 @@ func Ingest(ctx context.Context, st *store.Store, sourcePath string, opts Ingest
 	}
 	defer func() { _ = os.RemoveAll(filepath.Dir(snapshot.Root)) }()
 
-	extracted, err := Extract(snapshot.Root)
+	extracted, err := Extract(ctx, snapshot.Root)
 	if err != nil {
 		return Source{}, err
 	}

internal/slackdesktop/desktop_test.go

@@ -401,7 +401,7 @@ fs.writeFileSync(process.argv[1], v8.serialize(value));
 	blobPayload := append([]byte{0xff, 0x11, 0x02}, snappy.Encode(nil, serialized)...)
 	require.NoError(t, os.WriteFile(filepath.Join(root, "IndexedDB", "https_app.slack.com_0.indexeddb.blob", "1", "cd", "cd9a"), blobPayload, 0o600))
 
-	states, err := ExtractIndexedDBStates(root)
+	states, err := ExtractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Len(t, states, 1)
 	require.Equal(t, "T111", states[0].WorkspaceID)
@@ -1038,7 +1038,7 @@ func TestInspectIncludesSnapshotDerivedDesktopSummaries(t *testing.T) {
 	require.NoError(t, indexDB.Put([]byte("https_app.slack.com_0@1#objectStore-T111-U111"), []byte("A"), nil))
 	require.NoError(t, indexDB.Close())
 
-	source, err := Inspect(root)
+	source, err := Inspect(context.Background(), root)
 	require.NoError(t, err)
 	require.True(t, source.Available)
 	require.Equal(t, 1, source.Local.WorkspaceCount)

internal/slackdesktop/redux.go

@@ -120,12 +120,12 @@ func nodeAvailable() bool {
 	return err == nil
 }
 
-func ExtractIndexedDBStates(path string) ([]ReduxDecodedState, error) {
-	states, _, err := extractIndexedDBStates(path)
+func ExtractIndexedDBStates(ctx context.Context, path string) ([]ReduxDecodedState, error) {
+	states, _, err := extractIndexedDBStates(ctx, path)
 	return states, err
 }
 
-func extractIndexedDBStates(path string) ([]ReduxDecodedState, IndexedDBSummary, error) {
+func extractIndexedDBStates(ctx context.Context, path string) ([]ReduxDecodedState, IndexedDBSummary, error) {
 	refs, err := scanReduxBlobRefs(filepath.Join(path, indexedDBBlobDir))
 	if err != nil {
 		return nil, IndexedDBSummary{}, err
@@ -138,6 +138,9 @@ func extractIndexedDBStates(path string) ([]ReduxDecodedState, IndexedDBSummary,
 
 	byIdentity := map[string]ReduxDecodedState{}
 	for _, ref := range refs {
+		if err := ctx.Err(); err != nil {
+			return nil, summary, err
+		}
 		result := analyzeReduxBlob(ref.Path)
 		if !result.candidate {
 			if result.err != nil {
@@ -157,8 +160,11 @@ func extractIndexedDBStates(path string) ([]ReduxDecodedState, IndexedDBSummary,
 		if !summary.NodeAvailable {
 			continue
 		}
-		state, err := runReduxDecoder(result.payload)
+		state, err := runReduxDecoder(ctx, result.payload)
 		if err != nil {
+			if ctx.Err() != nil {
+				return nil, summary, ctx.Err()
+			}
 			summary.recordDecodeFailure(decodeErrorStage(err))
 			continue
 		}
@@ -464,12 +470,12 @@ type reduxBlobDecode struct {
 
 // decodeReduxBlob analyzes a blob file and, for supported candidates, decodes
 // it with Node.
-func decodeReduxBlob(blobPath string) reduxBlobDecode {
+func decodeReduxBlob(ctx context.Context, blobPath string) reduxBlobDecode {
 	result := analyzeReduxBlob(blobPath)
 	if !result.candidate || result.err != nil {
 		return result
 	}
-	state, err := runReduxDecoder(result.payload)
+	state, err := runReduxDecoder(ctx, result.payload)
 	if err != nil {
 		result.payload = nil
 		result.err = err
@@ -563,7 +569,7 @@ func isKnownBlinkVersion(version byte) bool {
 	return version >= minWireVersionWindow && version <= maxBlinkEnvelopeVersion
 }
 
-func runReduxDecoder(payload []byte) (ReduxDecodedState, error) {
+func runReduxDecoder(ctx context.Context, payload []byte) (ReduxDecodedState, error) {
 	tempFile, err := os.CreateTemp("", "slacrawl-redux-*.bin")
 	if err != nil {
 		return ReduxDecodedState{}, newReduxDecodeError("temp", err)
@@ -578,9 +584,12 @@ func runReduxDecoder(payload []byte) (ReduxDecodedState, error) {
 		return ReduxDecodedState{}, newReduxDecodeError("temp", err)
 	}
 
-	cmd := exec.Command("node", "-e", reduxDecoderScript, tempPath) //nolint:gosec // Node decodes a temporary V8 payload copied from the Slack data directory.
+	cmd := exec.CommandContext(ctx, "node", "-e", reduxDecoderScript, tempPath) //nolint:gosec // Node decodes a temporary V8 payload copied from the Slack data directory.
 	output, err := cmd.Output()
 	if err != nil {
+		if ctx.Err() != nil {
+			return ReduxDecodedState{}, ctx.Err()
+		}
 		return ReduxDecodedState{}, newReduxDecodeError("node", err)
 	}
 

internal/slackdesktop/redux_decode_test.go

@@ -9,6 +9,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/golang/snappy"
 	"github.com/stretchr/testify/require"
@@ -89,7 +90,7 @@ func decodeFixtureBlob(t *testing.T, payload []byte) reduxBlobDecode {
 	t.Helper()
 	blobPath := filepath.Join(t.TempDir(), "redux.blob")
 	require.NoError(t, os.WriteFile(blobPath, payload, 0o600))
-	return decodeReduxBlob(blobPath)
+	return decodeReduxBlob(context.Background(), blobPath)
 }
 
 func TestDecodeReduxBlobBareV15(t *testing.T) {
@@ -181,12 +182,42 @@ func TestLocateInnerV8Payload(t *testing.T) {
 	}
 }
 
+func TestExtractIndexedDBStatesCancelsHungDecoder(t *testing.T) {
+	requireNode(t)
+
+	orig := reduxDecoderScript
+	t.Cleanup(func() { reduxDecoderScript = orig })
+	reduxDecoderScript = "setTimeout(function () {}, 1e9)"
+
+	root := t.TempDir()
+	writeBlob(t, root, "hang-a", serializeReduxFixture(t, "T111", "U111"))
+	writeBlob(t, root, "hang-b", serializeReduxFixture(t, "T222", "U222"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() {
+		_, _, err := extractIndexedDBStates(ctx, root)
+		done <- err
+	}()
+
+	time.Sleep(80 * time.Millisecond)
+	cancel()
+
+	select {
+	case err := <-done:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(2 * time.Second):
+		t.Fatal("extractIndexedDBStates ignored context cancel for a hung node decoder")
+	}
+}
+
 func TestExtractIndexedDBStatesMixedV15AndV16(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "v15", serializeReduxFixture(t, "T111", "U111"))
 	writeBlob(t, root, "v16", snappyWrap(blink21Wrap(asWireFormatV16(serializeReduxFixture(t, "T222", "U222")))))
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Len(t, states, 2)
 	require.Equal(t, "T111", states[0].WorkspaceID)
@@ -207,7 +238,7 @@ func TestExtractIndexedDBStatesRetainsValidStatesWhenACandidateIsCorrupt(t *test
 	// Recognized V8 v15 header with a truncated payload: node must reject it.
 	writeBlob(t, root, "truncated", []byte{0xff, v8WireFormatV15, 0x6f})
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Len(t, states, 1)
 	require.Equal(t, "T111", states[0].WorkspaceID)
@@ -224,7 +255,7 @@ func TestExtractIndexedDBStatesIgnoresUnrelatedBlobFiles(t *testing.T) {
 	writeBlob(t, root, "valid", serializeReduxFixture(t, "T111", "U111"))
 	writeBlob(t, root, "unrelated", []byte("exported attachment bytes, not a redux state"))
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Len(t, states, 1)
 	require.Equal(t, 2, summary.BlobFileCount)
@@ -241,7 +272,7 @@ func TestExtractIndexedDBStatesUnreadableFileIsNotACandidate(t *testing.T) {
 	require.NoError(t, os.Chmod(unreadable, 0o000))
 	t.Cleanup(func() { _ = os.Chmod(unreadable, 0o600) })
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Len(t, states, 1)
 	require.Equal(t, 2, summary.BlobFileCount)
@@ -255,7 +286,7 @@ func TestExtractIndexedDBStatesNoCandidatesIsSuccessful(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "unrelated", []byte("not a redux blob"))
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Empty(t, states)
 	require.Equal(t, 1, summary.BlobFileCount)
@@ -267,7 +298,7 @@ func TestExtractIndexedDBStatesClassifiesMalformedSnappy(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "malformed", append([]byte{0xff, 0x11, 0x02}, []byte("%%%not-snappy%%%")...))
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Empty(t, states)
 	require.Equal(t, 1, summary.CandidateCount)
@@ -279,7 +310,7 @@ func TestExtractIndexedDBStatesReportsUnsupportedVersion(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "future", []byte{0xff, 17, 0x6f, 0x7b})
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Empty(t, states)
 	require.Equal(t, 1, summary.CandidateCount)
@@ -293,7 +324,7 @@ func TestExtractIndexedDBStatesNodeUnavailable(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "v15", []byte{0xff, v8WireFormatV15, 0x6f})
 
-	states, summary, err := extractIndexedDBStates(root)
+	states, summary, err := extractIndexedDBStates(context.Background(), root)
 	require.NoError(t, err)
 	require.Empty(t, states)
 	require.False(t, summary.NodeAvailable)
@@ -310,7 +341,7 @@ func TestInspectReportsAllCandidateFailureWithoutError(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "truncated", []byte{0xff, v8WireFormatV15, 0x6f})
 
-	source, err := Inspect(root)
+	source, err := Inspect(context.Background(), root)
 	require.NoError(t, err)
 	require.True(t, source.IndexedDB.NodeAvailable)
 	require.Equal(t, 1, source.IndexedDB.BlobFileCount)
@@ -368,7 +399,7 @@ func TestDoctorJSONIncludesIndexedDBDiagnostics(t *testing.T) {
 	root := t.TempDir()
 	writeBlob(t, root, "malformed", append([]byte{0xff, 0x11, 0x02}, []byte("supersecretpayload")...))
 
-	source, err := Inspect(root)
+	source, err := Inspect(context.Background(), root)
 	require.NoError(t, err)
 
 	rendered, err := json.Marshal(source)