diff --git a/cms/djangoapps/contentstore/api/views/course_import.py b/cms/djangoapps/contentstore/api/views/course_import.py
index f2f8f168df50..17152941a922 100644
--- a/cms/djangoapps/contentstore/api/views/course_import.py
+++ b/cms/djangoapps/contentstore/api/views/course_import.py
@@ -3,14 +3,12 @@
 """
 
 
-import base64
 import logging
 import os
+from uuid import uuid4
 
-from django.conf import settings
 from django.core.files import File
 from edx_django_utils.monitoring import set_custom_attribute, set_custom_attributes_for_course_key
-from path import Path as path
 from rest_framework import status
 from rest_framework.exceptions import AuthenticationFailed
 from rest_framework.generics import GenericAPIView
@@ -18,7 +16,12 @@
 from user_tasks.models import UserTaskStatus
 
 from cms.djangoapps.contentstore.storage import course_import_export_storage
-from cms.djangoapps.contentstore.tasks import CourseImportTask, import_olx
+from cms.djangoapps.contentstore.tasks import (
+    CourseImportTask,
+    course_import_working_dir,
+    import_olx,
+    remove_course_import_working_dir,
+)
 from cms.djangoapps.contentstore.utils import IMPORTABLE_FILE_TYPES
 from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, view_auth_classes
 
@@ -131,22 +134,26 @@ def post(self, request, course_key):
                     developer_message='Parameter in the wrong format',
                     error_code='internal_error',
                 )
-            course_dir = path(settings.GITHUB_REPO_ROOT) / base64.urlsafe_b64encode(
-                repr(course_key).encode('utf-8')
-            ).decode('utf-8')
+            # Staging directory private to this upload, so that a concurrent import of
+            # the same course cannot overwrite this archive or delete it on its way out.
+            course_dir = course_import_working_dir(course_key, f'upload-{uuid4().hex}')
             temp_filepath = course_dir / filename
-            if not course_dir.isdir():
-                os.mkdir(course_dir)
-
-            log.debug(f'importing course to {temp_filepath}')
-            with open(temp_filepath, "wb+") as temp_file:
-                for chunk in request.FILES['course_data'].chunks():
-                    temp_file.write(chunk)
-
-            log.info("Course import %s: Upload complete", course_key)
-            with open(temp_filepath, 'rb') as local_file:
-                django_file = File(local_file)
-                storage_path = course_import_export_storage.save('olx_import/' + filename, django_file)
+            os.makedirs(course_dir, exist_ok=True)
+
+            try:
+                log.debug(f'importing course to {temp_filepath}')
+                with open(temp_filepath, "wb+") as temp_file:
+                    for chunk in request.FILES['course_data'].chunks():
+                        temp_file.write(chunk)
+
+                log.info("Course import %s: Upload complete", course_key)
+                with open(temp_filepath, 'rb') as local_file:
+                    django_file = File(local_file)
+                    storage_path = course_import_export_storage.save('olx_import/' + filename, django_file)
+            finally:
+                # The archive now lives in storage; the import task downloads it into its
+                # own working directory, so this staging copy is no longer needed.
+                remove_course_import_working_dir(course_dir)
 
             async_result = import_olx.delay(
                 request.user.id, str(course_key), storage_path, filename, request.LANGUAGE_CODE)
diff --git a/cms/djangoapps/contentstore/tasks.py b/cms/djangoapps/contentstore/tasks.py
index 73f111ff8462..ab6f78d21782 100644
--- a/cms/djangoapps/contentstore/tasks.py
+++ b/cms/djangoapps/contentstore/tasks.py
@@ -522,6 +522,36 @@ def sync_discussion_settings(course_key, user):
         LOGGER.info(f'Course import {course.id}: DiscussionsConfiguration sync failed: {exc}')
 
 
+def course_import_working_dir(courselike_key, unique_id):
+    """
+    Return a private scratch directory for one course import.
+
+    ``unique_id`` must be unique per import (a task id, an upload id, ...). Every
+    import needs its own directory: they used to share a single directory derived
+    from the course key alone, so two imports of the same course would overwrite
+    each other's archive, and whichever import finished first deleted the other's
+    extracted OLX mid-run.
+    """
+    subdir = base64.urlsafe_b64encode(repr(courselike_key).encode('utf-8')).decode('utf-8')
+    return path(settings.GITHUB_REPO_ROOT) / subdir / str(unique_id)
+
+
+def remove_course_import_working_dir(course_dir):
+    """
+    Delete a working directory created by :func:`course_import_working_dir`.
+
+    Also drops the per-course parent directory once the last import using it is
+    gone. Both steps tolerate the directory already being missing, so that
+    cleanup never masks the error that triggered it.
+    """
+    shutil.rmtree(course_dir, ignore_errors=True)
+    try:
+        os.rmdir(os.path.dirname(course_dir))
+    except OSError:
+        # Still holds another import's working directory, or is already gone.
+        pass
+
+
 @shared_task(base=CourseImportTask, bind=True)
 # Note: The decorator @set_code_owner_attribute cannot be used here because the UserTaskMixin
 #   does stack inspection and can't handle additional decorators.
@@ -538,8 +568,7 @@ def import_olx(self, user_id, course_key_string, archive_path, archive_name, lan
     self.status.set_state(current_step)
 
     data_root = path(settings.GITHUB_REPO_ROOT)
-    subdir = base64.urlsafe_b64encode(repr(courselike_key).encode('utf-8')).decode('utf-8')
-    course_dir = data_root / subdir
+    course_dir = course_import_working_dir(courselike_key, self.request.id)
 
     def validate_user():
         """Validate if the user exists otherwise log error. """
@@ -647,8 +676,7 @@ def get_dir_for_filename(directory, filename):
         LOGGER.info(f'{log_prefix}: unpacking step started')
 
         temp_filepath = course_dir / get_valid_filename(archive_name)
-        if not course_dir.isdir():
-            os.mkdir(course_dir)
+        os.makedirs(course_dir, exist_ok=True)
 
         LOGGER.info(f'{log_prefix}: importing course to {temp_filepath}')
 
@@ -684,9 +712,8 @@ def read_chunk():
                 LOGGER.info(f'{log_prefix}: entrance exam milestone content reference has been removed')
     # Send errors to client with stage at which error occurred.
     except Exception as exception:  # pylint: disable=broad-except
-        if course_dir.isdir():
-            shutil.rmtree(course_dir)
-            LOGGER.info(f'{log_prefix}: Temp data cleared')
+        remove_course_import_working_dir(course_dir)
+        LOGGER.info(f'{log_prefix}: Temp data cleared')
 
         self.status.fail(UserErrors.UNKNOWN_ERROR_IN_UNPACKING)
         LOGGER.exception(f'{log_prefix}: Unknown error while unpacking', exc_info=True)
@@ -742,9 +769,8 @@ def read_chunk():
     except Exception as exception:  # pylint: disable=broad-except
         handle_course_import_exception(courselike_key, exception, self.status, known=False)
     finally:
-        if course_dir.isdir():
-            shutil.rmtree(course_dir)
-            LOGGER.info(f'{log_prefix}: Temp data cleared')
+        remove_course_import_working_dir(course_dir)
+        LOGGER.info(f'{log_prefix}: Temp data cleared')
 
         if self.status.state == 'Updating' and is_course:
             # Reload the course so we have the latest state
diff --git a/cms/djangoapps/contentstore/views/import_export.py b/cms/djangoapps/contentstore/views/import_export.py
index e4cc5a04cd17..ea5451c6b86a 100644
--- a/cms/djangoapps/contentstore/views/import_export.py
+++ b/cms/djangoapps/contentstore/views/import_export.py
@@ -4,12 +4,11 @@
 """
 
 
-import base64
+import hashlib
 import json
 import logging
 import os
 import re
-import shutil
 from wsgiref.util import FileWrapper
 
 from django.conf import settings
@@ -28,7 +27,6 @@
 from opaque_keys.edx.keys import CourseKey
 from opaque_keys.edx.locator import LibraryLocator
 from openedx_authz.constants.permissions import COURSES_EXPORT_COURSE, COURSES_IMPORT_COURSE
-from path import Path as path
 from storages.backends.s3boto3 import S3Boto3Storage
 from user_tasks.conf import settings as user_tasks_settings
 from user_tasks.models import UserTaskArtifact, UserTaskStatus
@@ -42,7 +40,15 @@
 from xmodule.modulestore.django import modulestore  # pylint: disable=wrong-import-order
 
 from ..storage import course_import_export_storage
-from ..tasks import CourseExportTask, CourseImportTask, create_export_tarball, export_olx, import_olx
+from ..tasks import (
+    CourseExportTask,
+    CourseImportTask,
+    course_import_working_dir,
+    create_export_tarball,
+    export_olx,
+    import_olx,
+    remove_course_import_working_dir,
+)
 from ..utils import IMPORTABLE_FILE_TYPES, get_export_url, get_import_url, reverse_course_url
 
 __all__ = [
@@ -110,15 +116,28 @@ def _save_request_status(request, key, status):
     request.session.save()
 
 
+def _upload_id(request, filename):
+    """
+    Identify one chunked upload, so that its staging directory is not shared.
+
+    Chunks of a single upload all have to land in the same directory, so the id
+    is derived from the session and the file name rather than being random.
+    """
+    session_key = request.session.session_key or ''
+    digest = hashlib.sha256(f'{session_key}:{filename}'.encode()).hexdigest()
+    return f'upload-{digest[:32]}'
+
+
 def _write_chunk(request, courselike_key):  # pylint: disable=too-many-statements
     """
     Write the OLX file data chunk from the given request to the local filesystem.
     """
-    # Upload .tar.gz or .zip to local filesystem for one-server installations not using S3 or Swift
-    data_root = path(settings.GITHUB_REPO_ROOT)
-    subdir = base64.urlsafe_b64encode(repr(courselike_key).encode('utf-8')).decode('utf-8')
-    course_dir = data_root / subdir
     filename = request.FILES['course-data'].name
+    # Upload .tar.gz or .zip to local filesystem for one-server installations not using S3 or Swift.
+    # The staging directory is private to this upload: it has to survive between chunks of the same
+    # upload, but must not be shared with another author's upload or with a running import task,
+    # which would delete it out from under us on its way out.
+    course_dir = course_import_working_dir(courselike_key, _upload_id(request, filename))
     set_custom_attributes_for_course_key(courselike_key)
     current_step = 'Uploading'
 
@@ -139,8 +158,7 @@ def error_response(message, status, stage):
             return error_response(error_message, 415, 0)
 
         temp_filepath = course_dir / filename
-        if not course_dir.isdir():
-            os.mkdir(course_dir)
+        os.makedirs(course_dir, exist_ok=True)
 
         logging.info(f'Course import {courselike_key}: importing course to {temp_filepath}')
 
@@ -207,15 +225,17 @@ def error_response(message, status, stage):
         with open(temp_filepath, 'rb') as local_file:
             django_file = File(local_file)
             storage_path = course_import_export_storage.save('olx_import/' + filename, django_file)
+        # The archive now lives in storage; the import task downloads it into its own
+        # working directory, so this staging copy is no longer needed.
+        remove_course_import_working_dir(course_dir)
         import_olx.delay(
             request.user.id, str(courselike_key), storage_path, filename, request.LANGUAGE_CODE)
 
     # Send errors to client with stage at which error occurred.
     except Exception as exception:  # pylint: disable=broad-except
         _save_request_status(request, courselike_string, -1)
-        if course_dir.isdir():
-            shutil.rmtree(course_dir)
-            log.info("Course import %s: Temp data cleared", courselike_key)
+        remove_course_import_working_dir(course_dir)
+        log.info("Course import %s: Temp data cleared", courselike_key)
 
         monitor_import_failure(courselike_key, current_step, exception=exception)
         log.exception(f'Course import {courselike_key}: error importing course.')
diff --git a/cms/djangoapps/contentstore/views/tests/test_import_export.py b/cms/djangoapps/contentstore/views/tests/test_import_export.py
index 9f0e7e0a9c35..9d072e4fe457 100644
--- a/cms/djangoapps/contentstore/views/tests/test_import_export.py
+++ b/cms/djangoapps/contentstore/views/tests/test_import_export.py
@@ -1,6 +1,7 @@
 """
 Unit tests for course import and export
 """
+import base64
 import copy
 import itertools
 import json
@@ -21,6 +22,7 @@
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.core.exceptions import SuspiciousOperation
+from django.core.files import File
 from django.core.files.storage import FileSystemStorage
 from django.test.utils import override_settings
 from milestones.tests.utils import MilestonesTestCaseMixin
@@ -30,11 +32,13 @@
 from rest_framework import status
 from rest_framework.test import APIClient
 from storages.backends.s3boto3 import S3Boto3Storage
-from user_tasks.models import UserTaskStatus
+from user_tasks.models import UserTaskArtifact, UserTaskStatus
 
 from cms.djangoapps.contentstore import errors as import_error
+from cms.djangoapps.contentstore import tasks
 from cms.djangoapps.contentstore.api.tests.base import BaseCourseViewTest
 from cms.djangoapps.contentstore.storage import course_import_export_storage
+from cms.djangoapps.contentstore.tasks import import_olx
 from cms.djangoapps.contentstore.tests.test_libraries import LibraryTestCase
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
 from cms.djangoapps.contentstore.utils import reverse_course_url
@@ -748,6 +752,123 @@ def test_import_status_response_is_not_cached(self, fmt):
         self.assertEqual(resp.headers['Cache-Control'], 'no-cache, no-store, must-revalidate')  # noqa: PT009
 
 
+@override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
+class ConcurrentImportTestCase(CourseTestCase):
+    """
+    Tests for two imports of the same course overlapping in time.
+
+    Regression tests for a bug where every import of a given course shared one
+    working directory, so whichever import finished first deleted the other's
+    archive and extracted OLX mid-run. The surviving task then failed with a
+    bare filesystem error (`[Errno 116] Stale file handle` on NFS-backed
+    ``GITHUB_REPO_ROOT``, `[Errno 2]` on a local disk).
+    """
+    ARCHIVE_NAME = 'course.tar.gz'
+
+    def setUp(self):
+        super().setUp()
+        self.content_dir = path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.content_dir, True)
+
+        source_dir = tempfile.mkdtemp(dir=self.content_dir)
+        os.makedirs(os.path.join(source_dir, 'course'))
+        with open(os.path.join(source_dir, 'course.xml'), 'w') as course_xml:
+            course_xml.write('<course url_name="2013_Spring" org="EDx" course="0.00x"/>')
+        with open(os.path.join(source_dir, 'course', '2013_Spring.xml'), 'w') as run_xml:
+            run_xml.write('<course></course>')
+
+        self.archive = os.path.join(self.content_dir, self.ARCHIVE_NAME)
+        with tarfile.open(self.archive, 'w:gz') as archive:
+            archive.add(source_dir, arcname='exported_course')
+
+        data_root = path(settings.GITHUB_REPO_ROOT)
+        if not data_root.isdir():
+            os.makedirs(data_root)
+        self.per_course_dir = data_root / base64.urlsafe_b64encode(
+            repr(self.course.id).encode('utf-8')
+        ).decode('utf-8')
+        self.addCleanup(shutil.rmtree, self.per_course_dir, True)
+
+    def stage_upload(self):
+        """Park a copy of the archive in storage, the way the upload view does."""
+        with open(self.archive, 'rb') as archive:
+            return course_import_export_storage.save(
+                'olx_import/' + self.ARCHIVE_NAME, File(archive)
+            )
+
+    def run_import(self, storage_path):
+        """Run one import_olx task to completion (celery is eager under test)."""
+        return import_olx.delay(
+            self.user.id, str(self.course.id), storage_path, self.ARCHIVE_NAME, 'en'
+        )
+
+    @staticmethod
+    def status_of(result):
+        """The UserTaskStatus recorded for a finished task."""
+        return UserTaskStatus.objects.get(task_id=result.id)
+
+    @staticmethod
+    def error_of(task_status):
+        """The error message a failed task showed the user, if any."""
+        artifact = UserTaskArtifact.objects.filter(status=task_status, name='Error').first()
+        return artifact.text if artifact else None
+
+    def test_concurrent_imports_of_same_course_both_succeed(self):
+        """
+        An import that finishes while a second one is still unpacking must not
+        disturb it.
+
+        Only the timing is simulated: the first import is run at the exact
+        moment the second one reaches its extraction step. Every filesystem
+        effect is real.
+        """
+        first_upload = self.stage_upload()
+        second_upload = self.stage_upload()
+        self.assertNotEqual(first_upload, second_upload)  # noqa: PT009
+
+        real_extractall = tasks.safe_extractall
+        observed = {}
+
+        def extract_with_a_concurrent_import(file_name, output_path):
+            """Runs as the second import extracts; the first import lands here."""
+            if not observed.get('fired'):
+                observed['fired'] = True  # the first import must not re-enter
+                observed['archive_before'] = os.path.exists(file_name)
+                observed['first_result'] = self.run_import(first_upload)
+                observed['archive_after'] = os.path.exists(file_name)
+                observed['dir_after'] = os.path.isdir(output_path)
+            return real_extractall(file_name, output_path)
+
+        with patch.object(tasks, 'safe_extractall', extract_with_a_concurrent_import):
+            second_result = self.run_import(second_upload)
+
+        first_status = self.status_of(observed['first_result'])
+        second_status = self.status_of(second_result)
+
+        self.assertEqual(first_status.state, UserTaskStatus.SUCCEEDED)  # noqa: PT009
+        # The first import's cleanup left the second import's files alone.
+        self.assertTrue(observed['archive_before'])  # noqa: PT009
+        self.assertTrue(observed['archive_after'])  # noqa: PT009
+        self.assertTrue(observed['dir_after'])  # noqa: PT009
+        self.assertEqual(  # noqa: PT009
+            second_status.state, UserTaskStatus.SUCCEEDED, self.error_of(second_status)
+        )
+        # ...and between them they left no scratch data behind.
+        self.assertFalse(os.path.exists(self.per_course_dir))  # noqa: PT009
+
+    def test_import_working_dirs_are_not_shared(self):
+        """Two imports of one course never get the same working directory."""
+        self.assertNotEqual(  # noqa: PT009
+            tasks.course_import_working_dir(self.course.id, 'task-1'),
+            tasks.course_import_working_dir(self.course.id, 'task-2'),
+        )
+        self.assertTrue(  # noqa: PT009
+            tasks.course_import_working_dir(self.course.id, 'task-1').startswith(
+                self.per_course_dir
+            )
+        )
+
+
 @override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
 @ddt.ddt
 class ExportTestCase(CourseTestCase):